3
0
mirror of https://git.kernel.org/pub/scm/network/wireless/iwd.git synced 2024-11-16 17:09:24 +01:00
iwd/tools/utils.py

586 lines
14 KiB
Python
Raw Normal View History

import os
import subprocess
import fcntl
import sys
import traceback
import shutil
import dbus
from datetime import datetime
from gi.repository import GLib
from weakref import WeakValueDictionary
from re import fullmatch
from time import sleep
from runner import RunnerCoreArgParse
class Process(subprocess.Popen):
processes = WeakValueDictionary()
testargs = RunnerCoreArgParse().parse_args()
def __new__(cls, *args, **kwargs):
obj = super().__new__(cls)
cls.processes[id(obj)] = obj
return obj
def __init__(self, args, namespace=None, outfile=None, env=None, check=False, cleanup=None):
self.write_fds = []
self.io_watch = None
self.cleanup = cleanup
self.verbose = False
self.out = ''
self.hup = False
self.killed = False
self.namespace = namespace
logfile = args[0]
if not shutil.which(args[0]):
raise Exception("%s is not found on system" % args[0])
if Process.is_verbose(args[0], log=False):
self.verbose = True
if namespace:
args = ['ip', 'netns', 'exec', namespace] + args
logfile += '-%s' % namespace
if outfile:
# outfile is only used by iwmon, in which case we don't want
# to append to an existing file.
self._append_outfile(outfile, append=False)
if self.testargs.log:
testdir = os.getcwd()
# Special case any processes started prior to a test
# (i.e. from testhome). Put these in the root log directory
if testdir == self.testargs.testhome:
testdir = '.'
else:
testdir = os.path.basename(testdir)
logfile = '%s/%s/%s' % (self.testargs.log, testdir, logfile)
self._append_outfile(logfile)
super().__init__(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
env=env, cwd=os.getcwd())
# Set as non-blocking so read() in the IO callback doesn't block forever
fl = fcntl.fcntl(self.stdout, fcntl.F_GETFL)
fcntl.fcntl(self.stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)
self.io_watch = GLib.io_add_watch(self.stdout, GLib.IO_IN |
GLib.IO_HUP | GLib.IO_ERR, self.process_io)
print("Starting process {}".format(self.args))
if check:
self.wait(10)
self.killed = True
if self.returncode != 0:
raise subprocess.CalledProcessError(returncode=self.returncode,
cmd=args)
@staticmethod
def is_verbose(process, log=True):
exclude = ['iwd-rtnl']
process = os.path.basename(process)
if Process.testargs is None:
return False
# every process is verbose when logging is enabled
if log and Process.testargs.log and process not in exclude:
return True
if process in Process.testargs.verbose:
return True
# Special case here to enable verbose output with valgrind running
if process == 'valgrind' and 'iwd' in Process.testargs.verbose:
return True
# Handle any regex matches
for item in Process.testargs.verbose:
try:
if fullmatch(item, process):
return True
except Exception as e:
print("%s is not a valid regex" % item)
return False
@classmethod
def get_all(cls):
return cls.processes.values()
@classmethod
def kill_all(cls):
for p in cls.processes.values():
if p.args[0] == 'dmesg':
continue
p.kill()
@staticmethod
def _write_io(instance, data, stdout=True):
for f in instance.write_fds:
for c in data:
f.write(c)
if c == '\n':
stamp = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S.%f")
f.write(stamp + ': ')
# Write out a separator so multiple process calls per
# test are easer to read.
if instance.hup:
f.write("Terminated: {}\n\n".format(instance.args))
f.flush()
if instance.verbose and stdout:
sys.__stdout__.write(data)
sys.__stdout__.flush()
@classmethod
def write_separators(cls, test, sep):
#
# There are either log running processes (cls.processes) or
# processes that have terminated already but a log file exists
# on disk. We still want the separators to show for both cases
# so after writing separators for running processes, also
# write them in any additional log files.
#
nowrite = []
for proc in cls.processes.values():
if proc.killed:
continue
cls._write_io(proc, sep, stdout=False)
nowrite.append(proc.args[0])
if cls.testargs.log:
logfiles = os.listdir('%s/%s' % (cls.testargs.log, test))
extra = list(set(logfiles) - set(nowrite))
for log in extra:
logfile = '%s/%s/%s' % (cls.testargs.log, test, log)
with open(logfile, 'a') as f:
f.write(sep)
f.close()
def process_io(self, source, condition):
if condition & GLib.IO_HUP:
self.hup = True
self.wait()
bt = self.out.partition("++++++++ backtrace ++++++++")
if bt[1]:
raise Exception(f"Process {self.args[0]} crashed!\n{bt[1] + bt[2]}")
data = source.read()
if not data:
return not self.hup
try:
data = data.decode('utf-8')
except:
return not self.hup
# Save data away in case the caller needs it (e.g. list_sta)
self.out += data
self._write_io(self, data)
return not self.hup
def _append_outfile(self, file, append=True):
gid = int(os.environ.get('SUDO_GID', os.getgid()))
uid = int(os.environ.get('SUDO_UID', os.getuid()))
dir = os.path.dirname(file)
if not os.path.exists(dir):
os.mkdir(dir)
os.chown(dir, uid, gid)
file = os.path.join(dir,file)
# If the out file exists, append. Useful for processes like
# hostapd_cli where it is called multiple times independently.
if os.path.isfile(file) and append:
mode = 'a'
else:
mode = 'w'
try:
f = open(file, mode)
except Exception as e:
traceback.print_exc()
sys.exit(0)
os.fchown(f.fileno(), uid, gid)
self.write_fds.append(f)
def wait_for_socket(self, socket, wait):
def _wait(socket):
if not os.path.exists(socket):
sleep(0.1)
return False
return True
Namespace.non_block_wait(_wait, wait, socket,
exception=Exception("Timed out waiting for %s" % socket))
def wait_for_service(self, ns, service, wait):
def _wait(ns, service):
if not ns._bus.name_has_owner(service):
sleep(0.1)
return False
return True
Namespace.non_block_wait(_wait, wait, ns, service,
exception=Exception("Timed out waiting for %s" % service))
# Wait for both process termination and HUP signal
def __wait(self, timeout):
try:
super().wait(timeout)
if not self.hup:
return False
return True
except:
return False
# Override wait() so it can do so non-blocking
def wait(self, timeout=10):
if timeout == None:
super().wait()
return
Namespace.non_block_wait(self.__wait, timeout, 1)
self._cleanup()
def _cleanup(self):
if self.cleanup:
self.cleanup()
self.write_fds = []
if self.io_watch:
GLib.source_remove(self.io_watch)
self.io_watch = None
self.cleanup = None
self.killed = True
# Override kill()
def kill(self, force=False):
if self.killed:
return
print("Killing process {}".format(self.args))
if force:
super().kill()
else:
self.terminate()
try:
self.wait(timeout=15)
except:
print("Process %s did not complete in 15 seconds!" % self.args[0])
super().kill()
self._cleanup()
def __str__(self):
return str(self.args) + '\n'
dbus_count = 0
# Partial DBus config. The remainder (<listen>) will be filled in for each
# namespace that is created so each individual dbus-daemon has its own socket
# and address.
dbus_config = '''
<!DOCTYPE busconfig PUBLIC \
"-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN" \
"http://www.freedesktop.org/standards/dbus/1.0/\
busconfig.dtd\">
<busconfig>
<type>system</type>
<limit name=\"reply_timeout\">2147483647</limit>
<auth>ANONYMOUS</auth>
<allow_anonymous/>
<policy context=\"default\">
<allow user=\"*\"/>
<allow own=\"*\"/>
<allow send_type=\"method_call\"/>
<allow send_type=\"signal\"/>
<allow send_type=\"method_return\"/>
<allow send_type=\"error\"/>
<allow receive_type=\"method_call\"/>
<allow receive_type=\"signal\"/>
<allow receive_type=\"method_return\"/>
<allow receive_type=\"error\"/>
<allow send_destination=\"*\" eavesdrop=\"true\"/>
<allow eavesdrop=\"true\"/>
</policy>
'''
class Namespace:
def __init__(self, args, name, radios):
self.dbus_address = None
self.name = name
self.radios = radios
self.args = args
Process(['ip', 'netns', 'add', name]).wait()
for r in radios:
r.set_namespace(self)
self.start_dbus()
def reset(self):
self._bus = None
for r in self.radios:
r._radio = None
self.radios = []
Process.kill_all()
def __del__(self):
if self.name:
print("Removing namespace %s" % self.name)
Process(['ip', 'netns', 'del', self.name]).wait()
def get_bus(self):
return self._bus
def start_process(self, args, env=None, **kwargs):
if not env:
env = os.environ.copy()
if hasattr(self, "dbus_address"):
# In case this process needs DBus...
env['DBUS_SYSTEM_BUS_ADDRESS'] = self.dbus_address
return Process(args, namespace=self.name, env=env, **kwargs)
def stop_process(self, p, force=False):
p.kill(force)
def _is_running(self, pid):
try:
os.kill(pid, 0)
except OSError:
return False
return True
def is_process_running(self, process):
for p in Process.get_all():
# Namespace processes are actually started by 'ip' where
# the actual process name is at index 4 of the arguments.
idx = 0 if not p.namespace else 4
if p.namespace == self.name and p.args[idx] == process:
# The process object exists, but make sure its
# actually running.
return self._is_running(p.pid)
return False
def _cleanup_dbus(self):
try:
os.remove(self.dbus_address.split('=')[1])
except:
pass
os.remove(self.dbus_cfg)
def start_dbus(self):
global dbus_count
self.dbus_address = 'unix:path=/tmp/dbus%d' % dbus_count
self.dbus_cfg = '/tmp/dbus%d.conf' % dbus_count
dbus_count += 1
with open(self.dbus_cfg, 'w+') as f:
f.write(dbus_config)
f.write('<listen>%s</listen>\n' % self.dbus_address)
f.write('</busconfig>\n')
p = self.start_process(['dbus-daemon', '--config-file=%s' % self.dbus_cfg],
cleanup=self._cleanup_dbus)
p.wait_for_socket(self.dbus_address.split('=')[1], 5)
self._bus = dbus.bus.BusConnection(address_or_type=self.dbus_address)
def start_iwd(self, config_dir = '/tmp', storage_dir = '/tmp/iwd',
developer_mode = True):
args = []
iwd_radios = ','.join([r.name for r in self.radios if r.use == 'iwd'])
if self.args.valgrind:
args.extend(['valgrind', '--leak-check=full', '--track-origins=yes',
'--show-leak-kinds=all',
'--log-file=/tmp/valgrind.log.%p'])
args.append('iwd')
if developer_mode:
args.append('-E')
if iwd_radios != '':
args.extend(['-p', iwd_radios])
if Process.is_verbose(args[0]):
args.append('-d')
env = os.environ.copy()
env['CONFIGURATION_DIRECTORY'] = config_dir
env['STATE_DIRECTORY'] = storage_dir
if Process.is_verbose('iwd-dhcp'):
env['IWD_DHCP_DEBUG'] = '1'
if Process.is_verbose('iwd-tls'):
env['IWD_TLS_DEBUG'] = '1'
if Process.is_verbose('iwd-acd'):
env['IWD_ACD_DEBUG'] = '1'
if Process.is_verbose('iwd-rtnl'):
env['IWD_RTNL_DEBUG'] = '1'
if Process.is_verbose('iwd-sae'):
env['IWD_SAE_DEBUG'] = '1'
proc = self.start_process(args, env=env)
proc.wait_for_service(self, 'net.connman.iwd', 20)
return proc
@staticmethod
def non_block_wait(func, timeout, *args, exception=True):
'''
Convenience function for waiting in a non blocking
manor using GLibs context iteration i.e. does not block
the main loop while waiting.
'func' will be called at least once and repeatedly until
either it returns success, throws an exception, or the
'timeout' expires.
'timeout' is the ultimate timeout in seconds
'*args' will be passed to 'func'
If 'exception' is an Exception type it will be raised.
If 'exception' is True a generic TimeoutError will be raised.
Any other value will not result in an exception.
'''
# Simple class for signaling the wait timeout
class Bool:
def __init__(self, value):
self.value = value
def wait_timeout_cb(done):
done.value = True
return False
mainloop = GLib.MainLoop()
done = Bool(False)
timeout = GLib.timeout_add_seconds(timeout, wait_timeout_cb, done)
context = mainloop.get_context()
while True:
try:
ret = func(*args)
if ret:
if not done.value:
GLib.source_remove(timeout)
return ret
except Exception as e:
if not done.value:
GLib.source_remove(timeout)
raise e
if done.value == True:
if isinstance(exception, Exception):
raise exception
elif type(exception) == bool and exception:
raise TimeoutError("Timeout on non_block_wait")
else:
return
context.iteration(may_block=True)
def __str__(self):
ret = 'Namespace: %s\n' % self.name
ret += 'Processes:\n'
for p in Process.get_all():
ret += '\t%s' % str(p)
ret += 'Radios:\n'
if len(self.radios) > 0:
for r in self.radios:
ret += '\t%s\n' % str(r)
else:
ret += '\tNo Radios\n'
ret += 'DBus Address: %s\n' % self.dbus_address
ret += '===================================================\n\n'
return ret
class BarChart():
def __init__(self, height=10, max_width=80):
self._height = height
self._max_width = max_width
self._values = []
self._max_value = 0
self._min_value = 0
def add_value(self, value):
if len(self._values) == 0:
self._max_value = int(1.01 * value)
self._min_value = int(0.99 * value)
elif value > self._max_value:
self._max_value = int(1.01 * value)
elif value < self._min_value:
self._min_value = int(0.99 * value)
self._values.append(value)
def _value_to_stars(self, value):
# Need to scale value (range of min_value -> max_value) to
# a range of 0 -> height
#
# Scaled = ((value - min_value) / ( max_value - min_value)) * (Height - 0) + 0
return int(((value - self._min_value) /
(self._max_value - self._min_value)) * self._height)
def __str__(self):
# Need to map value from range 0 - self._height
ret = ''
for i, value in enumerate(self._values):
stars = self._value_to_stars(value)
ret += '[%3u] ' % i + '%-10s' % ('*' * stars) + '\t\t\t%d\n' % value
ret += '\n'
return ret