271
## Classes for commands.
273
# Abstract classes first
274
class Command(object):
275
"""Abstract class for commands"""
276
def run(self, mandos, clients):
277
"""Normal commands should implement run_on_one_client(), but
278
commands which want to operate on all clients at the same time
279
can override this run() method instead."""
281
for client in clients:
282
self.run_on_one_client(client)
284
class PrintCmd(Command):
285
"""Abstract class for commands printing client details"""
286
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
287
"Created", "Interval", "Host", "KeyID",
288
"Fingerprint", "CheckerRunning", "LastEnabled",
289
"ApprovalPending", "ApprovedByDefault",
290
"LastApprovalRequest", "ApprovalDelay",
291
"ApprovalDuration", "Checker", "ExtendedTimeout",
292
"Expires", "LastCheckerStatus")
293
def run(self, mandos, clients):
294
print(self.output(clients))
296
class PropertyCmd(Command):
297
"""Abstract class for Actions for setting one client property"""
298
def run_on_one_client(self, client):
299
"""Set the Client's D-Bus property"""
300
client.Set(client_interface, self.property, self.value_to_set,
301
dbus_interface=dbus.PROPERTIES_IFACE)
303
class ValueArgumentMixIn(object):
304
"""Mixin class for commands taking a value as argument"""
305
def __init__(self, value):
306
self.value_to_set = value
308
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
309
"""Mixin class for commands taking a value argument as
312
def value_to_set(self):
315
def value_to_set(self, value):
316
"""When setting, convert value to a datetime.timedelta"""
317
self._vts = string_to_delta(value).total_seconds() * 1000
319
# Actual (non-abstract) command classes
321
class PrintTableCmd(PrintCmd):
322
def __init__(self, verbose=False):
323
self.verbose = verbose
325
def output(self, clients):
327
keywords = self.all_keywords
329
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
330
return str(self.TableOfClients(clients.values(), keywords))
332
class TableOfClients(object):
335
"Enabled": "Enabled",
336
"Timeout": "Timeout",
337
"LastCheckedOK": "Last Successful Check",
338
"LastApprovalRequest": "Last Approval Request",
339
"Created": "Created",
340
"Interval": "Interval",
342
"Fingerprint": "Fingerprint",
344
"CheckerRunning": "Check Is Running",
345
"LastEnabled": "Last Enabled",
346
"ApprovalPending": "Approval Is Pending",
347
"ApprovedByDefault": "Approved By Default",
348
"ApprovalDelay": "Approval Delay",
349
"ApprovalDuration": "Approval Duration",
350
"Checker": "Checker",
351
"ExtendedTimeout": "Extended Timeout",
352
"Expires": "Expires",
353
"LastCheckerStatus": "Last Checker Status",
356
def __init__(self, clients, keywords, tableheaders=None):
357
self.clients = clients
358
self.keywords = keywords
359
if tableheaders is not None:
360
self.tableheaders = tableheaders
363
return "\n".join(self.rows())
365
if sys.version_info.major == 2:
366
__unicode__ = __str__
368
return str(self).encode(locale.getpreferredencoding())
371
format_string = self.row_formatting_string()
372
rows = [self.header_line(format_string)]
373
rows.extend(self.client_line(client, format_string)
374
for client in self.clients)
377
def row_formatting_string(self):
378
"Format string used to format table rows"
379
return " ".join("{{{key}:{width}}}".format(
380
width=max(len(self.tableheaders[key]),
381
*(len(self.string_from_client(client, key))
382
for client in self.clients)),
384
for key in self.keywords)
386
def string_from_client(self, client, key):
387
return self.valuetostring(client[key], key)
390
def valuetostring(value, keyword):
391
if isinstance(value, dbus.Boolean):
392
return "Yes" if value else "No"
393
if keyword in ("Timeout", "Interval", "ApprovalDelay",
394
"ApprovalDuration", "ExtendedTimeout"):
395
return milliseconds_to_string(value)
398
def header_line(self, format_string):
399
return format_string.format(**self.tableheaders)
401
def client_line(self, client, format_string):
402
return format_string.format(
403
**{key: self.string_from_client(client, key)
404
for key in self.keywords})
408
class DumpJSONCmd(PrintCmd):
409
def output(self, clients):
410
data = {client["Name"]:
411
{key: self.dbus_boolean_to_bool(client[key])
412
for key in self.all_keywords}
413
for client in clients.values()}
414
return json.dumps(data, indent=4, separators=(',', ': '))
416
def dbus_boolean_to_bool(value):
417
if isinstance(value, dbus.Boolean):
421
class IsEnabledCmd(Command):
422
def run_on_one_client(self, client):
423
if self.is_enabled(client):
426
def is_enabled(self, client):
427
return client.Get(client_interface, "Enabled",
428
dbus_interface=dbus.PROPERTIES_IFACE)
430
class RemoveCmd(Command):
431
def run_on_one_client(self, client):
432
self.mandos.RemoveClient(client.__dbus_object_path__)
434
class ApproveCmd(Command):
435
def run_on_one_client(self, client):
436
client.Approve(dbus.Boolean(True),
437
dbus_interface=client_interface)
439
class DenyCmd(Command):
440
def run_on_one_client(self, client):
441
client.Approve(dbus.Boolean(False),
442
dbus_interface=client_interface)
444
class EnableCmd(PropertyCmd):
446
value_to_set = dbus.Boolean(True)
448
class DisableCmd(PropertyCmd):
450
value_to_set = dbus.Boolean(False)
452
class BumpTimeoutCmd(PropertyCmd):
453
property = "LastCheckedOK"
456
class StartCheckerCmd(PropertyCmd):
457
property = "CheckerRunning"
458
value_to_set = dbus.Boolean(True)
460
class StopCheckerCmd(PropertyCmd):
461
property = "CheckerRunning"
462
value_to_set = dbus.Boolean(False)
464
class ApproveByDefaultCmd(PropertyCmd):
465
property = "ApprovedByDefault"
466
value_to_set = dbus.Boolean(True)
468
class DenyByDefaultCmd(PropertyCmd):
469
property = "ApprovedByDefault"
470
value_to_set = dbus.Boolean(False)
472
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
475
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
478
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
481
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
484
class SetExtendedTimeoutCmd(PropertyCmd,
485
MillisecondsValueArgumentMixIn):
486
property = "ExtendedTimeout"
488
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
489
property = "Interval"
491
class SetApprovalDelayCmd(PropertyCmd,
492
MillisecondsValueArgumentMixIn):
493
property = "ApprovalDelay"
495
class SetApprovalDurationCmd(PropertyCmd,
496
MillisecondsValueArgumentMixIn):
497
property = "ApprovalDuration"
499
def has_actions(options):
500
return any((options.enable,
502
options.bump_timeout,
503
options.start_checker,
504
options.stop_checker,
507
options.checker is not None,
508
options.timeout is not None,
509
options.extended_timeout is not None,
510
options.interval is not None,
511
options.approved_by_default is not None,
512
options.approval_delay is not None,
513
options.approval_duration is not None,
514
options.host is not None,
515
options.secret is not None,
519
def add_command_line_options(parser):
520
parser.add_argument("--version", action="version",
521
version="%(prog)s {}".format(version),
522
help="show version number and exit")
523
parser.add_argument("-a", "--all", action="store_true",
524
help="Select all clients")
525
parser.add_argument("-v", "--verbose", action="store_true",
526
help="Print all fields")
527
parser.add_argument("-j", "--dump-json", action="store_true",
528
help="Dump client data in JSON format")
529
enable_disable = parser.add_mutually_exclusive_group()
530
enable_disable.add_argument("-e", "--enable", action="store_true",
531
help="Enable client")
532
enable_disable.add_argument("-d", "--disable",
534
help="disable client")
535
parser.add_argument("-b", "--bump-timeout", action="store_true",
536
help="Bump timeout for client")
537
start_stop_checker = parser.add_mutually_exclusive_group()
538
start_stop_checker.add_argument("--start-checker",
540
help="Start checker for client")
541
start_stop_checker.add_argument("--stop-checker",
543
help="Stop checker for client")
544
parser.add_argument("-V", "--is-enabled", action="store_true",
545
help="Check if client is enabled")
546
parser.add_argument("-r", "--remove", action="store_true",
547
help="Remove client")
548
parser.add_argument("-c", "--checker",
549
help="Set checker command for client")
550
parser.add_argument("-t", "--timeout",
551
help="Set timeout for client")
552
parser.add_argument("--extended-timeout",
553
help="Set extended timeout for client")
554
parser.add_argument("-i", "--interval",
555
help="Set checker interval for client")
556
approve_deny_default = parser.add_mutually_exclusive_group()
557
approve_deny_default.add_argument(
558
"--approve-by-default", action="store_true",
559
default=None, dest="approved_by_default",
560
help="Set client to be approved by default")
561
approve_deny_default.add_argument(
562
"--deny-by-default", action="store_false",
563
dest="approved_by_default",
564
help="Set client to be denied by default")
565
parser.add_argument("--approval-delay",
566
help="Set delay before client approve/deny")
567
parser.add_argument("--approval-duration",
568
help="Set duration of one client approval")
569
parser.add_argument("-H", "--host", help="Set host for client")
570
parser.add_argument("-s", "--secret",
571
type=argparse.FileType(mode="rb"),
572
help="Set password blob (file) for client")
573
approve_deny = parser.add_mutually_exclusive_group()
574
approve_deny.add_argument(
575
"-A", "--approve", action="store_true",
576
help="Approve any current client request")
577
approve_deny.add_argument("-D", "--deny", action="store_true",
578
help="Deny any current client request")
579
parser.add_argument("--check", action="store_true",
580
help="Run self-test")
581
parser.add_argument("client", nargs="*", help="Client name")
584
def commands_and_clients_from_options(options):
588
if options.dump_json:
589
commands.append(DumpJSONCmd())
592
commands.append(EnableCmd())
595
commands.append(DisableCmd())
597
if options.bump_timeout:
598
commands.append(BumpTimeoutCmd(options.bump_timeout))
600
if options.start_checker:
601
commands.append(StartCheckerCmd())
603
if options.stop_checker:
604
commands.append(StopCheckerCmd())
606
if options.is_enabled:
607
commands.append(IsEnabledCmd())
610
commands.append(RemoveCmd())
612
if options.checker is not None:
613
commands.append(SetCheckerCmd())
615
if options.timeout is not None:
616
commands.append(SetTimeoutCmd(options.timeout))
618
if options.extended_timeout:
620
SetExtendedTimeoutCmd(options.extended_timeout))
622
if options.interval is not None:
623
command.append(SetIntervalCmd(options.interval))
625
if options.approved_by_default is not None:
626
if options.approved_by_default:
627
command.append(ApproveByDefaultCmd())
629
command.append(DenyByDefaultCmd())
631
if options.approval_delay is not None:
632
command.append(SetApprovalDelayCmd(options.approval_delay))
634
if options.approval_duration is not None:
636
SetApprovalDurationCmd(options.approval_duration))
638
if options.host is not None:
639
command.append(SetHostCmd(options.host))
641
if options.secret is not None:
642
command.append(SetSecretCmd(options.secret))
645
commands.append(ApproveCmd())
648
commands.append(DenyCmd())
650
# If no command option has been given, show table of clients,
651
# optionally verbosely
653
commands.append(PrintTableCmd(verbose=options.verbose))
655
return commands, options.client
659
parser = argparse.ArgumentParser()
661
add_command_line_options(parser)
663
options = parser.parse_args()
665
if has_actions(options) and not (options.client or options.all):
441
def check_option_syntax(parser, options):
442
"""Apply additional restrictions on options, not expressible in
445
def has_commands(options, commands=None):
447
commands = (command.Enable,
450
command.StartChecker,
456
command.SetExtendedTimeout,
458
command.ApproveByDefault,
459
command.DenyByDefault,
460
command.SetApprovalDelay,
461
command.SetApprovalDuration,
466
return any(isinstance(cmd, commands)
467
for cmd in options.commands)
469
if has_commands(options) and not (options.client or options.all):
666
470
parser.error("Options require clients names or --all.")
667
if options.verbose and has_actions(options):
471
if options.verbose and has_commands(options):
668
472
parser.error("--verbose can only be used alone.")
669
if options.dump_json and (options.verbose
670
or has_actions(options)):
473
if (has_commands(options, (command.DumpJSON,))
474
and (options.verbose or len(options.commands) > 1)):
671
475
parser.error("--dump-json can only be used alone.")
672
if options.all and not has_actions(options):
476
if options.all and not has_commands(options):
673
477
parser.error("--all requires an action.")
674
if options.is_enabled and len(options.client) > 1:
478
if (has_commands(options, (command.IsEnabled,))
479
and len(options.client) > 1):
675
480
parser.error("--is-enabled requires exactly one client")
677
commands, clientnames = commands_and_clients_from_options(options)
680
bus = dbus.SystemBus()
681
mandos_dbus_objc = bus.get_object(busname, server_path)
682
except dbus.exceptions.DBusException:
683
log.critical("Could not connect to Mandos server")
686
mandos_serv = dbus.Interface(mandos_dbus_objc,
687
dbus_interface=server_interface)
688
mandos_serv_object_manager = dbus.Interface(
689
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
691
# Filter out log message from dbus module
692
dbus_logger = logging.getLogger("dbus.proxies")
693
class NullFilter(logging.Filter):
694
def filter(self, record):
696
dbus_filter = NullFilter()
698
dbus_logger.addFilter(dbus_filter)
699
mandos_clients = {path: ifs_and_props[client_interface]
700
for path, ifs_and_props in
701
mandos_serv_object_manager
702
.GetManagedObjects().items()
703
if client_interface in ifs_and_props}
704
except dbus.exceptions.DBusException as e:
705
log.critical("Failed to access Mandos server through D-Bus:"
709
# restore dbus logger
710
dbus_logger.removeFilter(dbus_filter)
712
# Compile dict of (clients: properties) to process
716
clients = {bus.get_object(busname, path): properties
717
for path, properties in mandos_clients.items()}
719
for name in clientnames:
720
for path, client in mandos_clients.items():
721
if client["Name"] == name:
722
client_objc = bus.get_object(busname, path)
723
clients[client_objc] = client
726
log.critical("Client not found on server: %r", name)
729
# Run all commands on clients
730
for command in commands:
731
command.run(mandos_serv, clients)
481
if (len(options.commands) > 1
482
and has_commands(options, (command.Remove,))
483
and not has_commands(options, (command.Deny,))):
484
parser.error("--remove can only be combined with --deny")
491
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
493
def get_managed_objects(self, busname, objectpath):
494
return self.call_method("GetManagedObjects", busname,
496
self.object_manager_iface)
498
properties_iface = "org.freedesktop.DBus.Properties"
500
def set_property(self, busname, objectpath, interface, key,
502
self.call_method("Set", busname, objectpath,
503
self.properties_iface, interface, key,
506
def call_method(self, methodname, busname, objectpath,
508
raise NotImplementedError()
510
class MandosBus(SystemBus):
511
busname_domain = "se.recompile"
512
busname = busname_domain + ".Mandos"
514
server_interface = busname_domain + ".Mandos"
515
client_interface = busname_domain + ".Mandos.Client"
518
def get_clients_and_properties(self):
519
managed_objects = self.get_managed_objects(
520
self.busname, self.server_path)
521
return {objpath: properties[self.client_interface]
522
for objpath, properties in managed_objects.items()
523
if self.client_interface in properties}
525
def set_client_property(self, objectpath, key, value):
526
return self.set_property(self.busname, objectpath,
527
self.client_interface, key,
530
def call_client_method(self, objectpath, method, *args):
531
return self.call_method(method, self.busname, objectpath,
532
self.client_interface, *args)
534
def call_server_method(self, method, *args):
535
return self.call_method(method, self.busname,
537
self.server_interface, *args)
539
class Error(Exception):
542
class ConnectFailed(Error):
546
class dbus_python_adapter:
548
class SystemBus(dbus.MandosBus):
549
"""Use dbus-python"""
551
def __init__(self, module=dbus_python):
552
self.dbus_python = module
553
self.bus = self.dbus_python.SystemBus()
555
@contextlib.contextmanager
556
def convert_exception(self, exception_class=dbus.Error):
559
except self.dbus_python.exceptions.DBusException as e:
560
# This does what "raise from" would do
561
exc = exception_class(*e.args)
565
def call_method(self, methodname, busname, objectpath,
567
proxy_object = self.get_object(busname, objectpath)
568
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
569
interface, methodname,
570
", ".join(repr(a) for a in args))
571
method = getattr(proxy_object, methodname)
572
with self.convert_exception():
573
with dbus_python_adapter.SilenceLogger(
575
value = method(*args, dbus_interface=interface)
576
return self.type_filter(value)
578
def get_object(self, busname, objectpath):
579
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
581
with self.convert_exception(dbus.ConnectFailed):
582
return self.bus.get_object(busname, objectpath)
584
def type_filter(self, value):
585
"""Convert the most bothersome types to Python types"""
586
if isinstance(value, self.dbus_python.Boolean):
588
if isinstance(value, self.dbus_python.ObjectPath):
590
# Also recurse into dictionaries
591
if isinstance(value, self.dbus_python.Dictionary):
592
return {self.type_filter(key):
593
self.type_filter(subval)
594
for key, subval in value.items()}
597
def set_client_property(self, objectpath, key, value):
599
if not isinstance(value, bytes):
600
value = value.encode("utf-8")
601
value = self.dbus_python.ByteArray(value)
602
return self.set_property(self.busname, objectpath,
603
self.client_interface, key,
607
"Simple context manager to silence a particular logger"
609
def __init__(self, loggername):
610
self.logger = logging.getLogger(loggername)
613
self.logger.addFilter(self.nullfilter)
615
class NullFilter(logging.Filter):
616
def filter(self, record):
619
nullfilter = NullFilter()
621
def __exit__(self, exc_type, exc_val, exc_tb):
622
self.logger.removeFilter(self.nullfilter)
624
class CachingBus(SystemBus):
625
"""A caching layer for dbus_python_adapter.SystemBus"""
627
def __init__(self, *args, **kwargs):
628
self.object_cache = {}
629
super(dbus_python_adapter.CachingBus,
630
self).__init__(*args, **kwargs)
632
def get_object(self, busname, objectpath):
634
return self.object_cache[(busname, objectpath)]
637
dbus_python_adapter.CachingBus,
638
self).get_object(busname, objectpath)
639
self.object_cache[(busname, objectpath)] = new_object
643
class pydbus_adapter:
644
class SystemBus(dbus.MandosBus):
645
def __init__(self, module=pydbus):
647
self.bus = self.pydbus.SystemBus()
649
@contextlib.contextmanager
650
def convert_exception(self, exception_class=dbus.Error):
653
except gi.repository.GLib.Error as e:
654
# This does what "raise from" would do
655
exc = exception_class(*e.args)
659
def call_method(self, methodname, busname, objectpath,
661
proxy_object = self.get(busname, objectpath)
662
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
663
interface, methodname,
664
", ".join(repr(a) for a in args))
665
method = getattr(proxy_object[interface], methodname)
666
with self.convert_exception():
669
def get(self, busname, objectpath):
670
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
672
with self.convert_exception(dbus.ConnectFailed):
673
if sys.version_info.major <= 2:
674
with warnings.catch_warnings():
675
warnings.filterwarnings(
676
"ignore", "", DeprecationWarning,
677
r"^xml\.etree\.ElementTree$")
678
return self.bus.get(busname, objectpath)
680
return self.bus.get(busname, objectpath)
682
def set_property(self, busname, objectpath, interface, key,
684
proxy_object = self.get(busname, objectpath)
685
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
686
objectpath, self.properties_iface, interface,
688
setattr(proxy_object[interface], key, value)
690
class CachingBus(SystemBus):
691
"""A caching layer for pydbus_adapter.SystemBus"""
693
def __init__(self, *args, **kwargs):
694
self.object_cache = {}
695
super(pydbus_adapter.CachingBus,
696
self).__init__(*args, **kwargs)
698
def get(self, busname, objectpath):
700
return self.object_cache[(busname, objectpath)]
702
new_object = (super(pydbus_adapter.CachingBus, self)
703
.get(busname, objectpath))
704
self.object_cache[(busname, objectpath)] = new_object
708
class dbussy_adapter:
709
class SystemBus(dbus.SystemBus):
712
def __init__(self, dbussy, ravel):
715
self.bus = ravel.system_bus()
717
@contextlib.contextmanager
718
def convert_exception(self, exception_class=dbus.Error):
721
except self.dbussy.DBusError as e:
722
# This does what "raise from" would do
723
exc = exception_class(*e.args)
727
def call_method(self, methodname, busname, objectpath,
729
proxy_object = self.get_object(busname, objectpath)
730
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
731
interface, methodname,
732
", ".join(repr(a) for a in args))
733
iface = proxy_object.get_interface(interface)
734
method = getattr(iface, methodname)
735
with self.convert_exception(dbus.Error):
736
value = method(*args)
737
# DBussy returns values either as an empty list or as a
738
# list of one element with the return value
740
return self.type_filter(value[0])
742
def get_object(self, busname, objectpath):
743
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
745
with self.convert_exception(dbus.ConnectFailed):
746
return self.bus[busname][objectpath]
748
def type_filter(self, value):
749
"""Convert the most bothersome types to Python types"""
750
# A D-Bus Variant value is represented as the Python type
751
# Tuple[dbussy.DBUS.Signature, Any]
752
if isinstance(value, tuple):
754
and isinstance(value[0],
755
self.dbussy.DBUS.Signature)):
756
return self.type_filter(value[1])
757
elif isinstance(value, self.dbussy.DBUS.ObjectPath):
759
# Also recurse into dictionaries
760
elif isinstance(value, dict):
761
return {self.type_filter(key):
762
self.type_filter(subval)
763
for key, subval in value.items()}
766
def set_property(self, busname, objectpath, interface, key,
768
proxy_object = self.get_object(busname, objectpath)
769
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
770
objectpath, self.properties_iface, interface,
773
# DBussy wants a Byte Array to be a sequence of
774
# values, not a byte string
776
setattr(proxy_object.get_interface(interface), key, value)
778
class MandosBus(SystemBus, dbus.MandosBus):
781
class CachingBus(MandosBus):
782
"""A caching layer for dbussy_adapter.MandosBus"""
784
def __init__(self, *args, **kwargs):
785
self.object_cache = {}
786
super(dbussy_adapter.CachingBus, self).__init__(*args,
789
def get_object(self, busname, objectpath):
791
return self.object_cache[(busname, objectpath)]
794
dbussy_adapter.CachingBus,
795
self).get_object(busname, objectpath)
796
self.object_cache[(busname, objectpath)] = new_object
800
def commands_from_options(options):
802
commands = list(options.commands)
804
def find_cmd(cmd, commands):
806
for i, c in enumerate(commands):
807
if isinstance(c, cmd):
811
# If command.Remove is present, move any instances of command.Deny
812
# to occur ahead of command.Remove.
813
index_of_remove = find_cmd(command.Remove, commands)
814
before_remove = commands[:index_of_remove]
815
after_remove = commands[index_of_remove:]
817
for cmd in after_remove:
818
if isinstance(cmd, command.Deny):
819
before_remove.append(cmd)
821
cleaned_after.append(cmd)
822
if cleaned_after != after_remove:
823
commands = before_remove + cleaned_after
825
# If no command option has been given, show table of clients,
826
# optionally verbosely
828
commands.append(command.PrintTable(verbose=options.verbose))
834
"""A namespace for command classes"""
837
"""Abstract base class for commands"""
839
def run(self, clients, bus=None):
840
"""Normal commands should implement run_on_one_client(),
841
but commands which want to operate on all clients at the same time can
842
override this run() method instead.
845
for client, properties in clients.items():
846
self.run_on_one_client(client, properties)
848
class IsEnabled(Base):
849
def run(self, clients, bus=None):
850
properties = next(iter(clients.values()))
851
if properties["Enabled"]:
856
def run_on_one_client(self, client, properties):
857
self.bus.call_client_method(client, "Approve", True)
860
def run_on_one_client(self, client, properties):
861
self.bus.call_client_method(client, "Approve", False)
864
def run(self, clients, bus):
865
for clientpath in frozenset(clients.keys()):
866
bus.call_server_method("RemoveClient", clientpath)
869
"""Abstract class for commands outputting client details"""
870
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
871
"Created", "Interval", "Host", "KeyID",
872
"Fingerprint", "CheckerRunning",
873
"LastEnabled", "ApprovalPending",
874
"ApprovedByDefault", "LastApprovalRequest",
875
"ApprovalDelay", "ApprovalDuration",
876
"Checker", "ExtendedTimeout", "Expires",
879
class DumpJSON(Output):
880
def run(self, clients, bus=None):
881
data = {properties["Name"]:
882
{key: properties[key]
883
for key in self.all_keywords}
884
for properties in clients.values()}
885
print(json.dumps(data, indent=4, separators=(",", ": ")))
887
class PrintTable(Output):
888
def __init__(self, verbose=False):
889
self.verbose = verbose
891
def run(self, clients, bus=None):
892
default_keywords = ("Name", "Enabled", "Timeout",
894
keywords = default_keywords
896
keywords = self.all_keywords
897
print(self.TableOfClients(clients.values(), keywords))
899
class TableOfClients:
902
"Enabled": "Enabled",
903
"Timeout": "Timeout",
904
"LastCheckedOK": "Last Successful Check",
905
"LastApprovalRequest": "Last Approval Request",
906
"Created": "Created",
907
"Interval": "Interval",
909
"Fingerprint": "Fingerprint",
911
"CheckerRunning": "Check Is Running",
912
"LastEnabled": "Last Enabled",
913
"ApprovalPending": "Approval Is Pending",
914
"ApprovedByDefault": "Approved By Default",
915
"ApprovalDelay": "Approval Delay",
916
"ApprovalDuration": "Approval Duration",
917
"Checker": "Checker",
918
"ExtendedTimeout": "Extended Timeout",
919
"Expires": "Expires",
920
"LastCheckerStatus": "Last Checker Status",
923
def __init__(self, clients, keywords):
924
self.clients = clients
925
self.keywords = keywords
928
return "\n".join(self.rows())
930
if sys.version_info.major == 2:
931
__unicode__ = __str__
934
return str(self).encode(
935
locale.getpreferredencoding())
938
format_string = self.row_formatting_string()
939
rows = [self.header_line(format_string)]
940
rows.extend(self.client_line(client, format_string)
941
for client in self.clients)
944
def row_formatting_string(self):
945
"Format string used to format table rows"
946
return " ".join("{{{key}:{width}}}".format(
947
width=max(len(self.tableheaders[key]),
948
*(len(self.string_from_client(client,
950
for client in self.clients)),
952
for key in self.keywords)
954
def string_from_client(self, client, key):
955
return self.valuetostring(client[key], key)
958
def valuetostring(cls, value, keyword):
959
if isinstance(value, bool):
960
return "Yes" if value else "No"
961
if keyword in ("Timeout", "Interval", "ApprovalDelay",
962
"ApprovalDuration", "ExtendedTimeout"):
963
return cls.milliseconds_to_string(value)
966
def header_line(self, format_string):
967
return format_string.format(**self.tableheaders)
969
def client_line(self, client, format_string):
970
return format_string.format(
971
**{key: self.string_from_client(client, key)
972
for key in self.keywords})
975
def milliseconds_to_string(ms):
976
td = datetime.timedelta(0, 0, 0, ms)
977
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
978
.format(days="{}T".format(td.days)
980
hours=td.seconds // 3600,
981
minutes=(td.seconds % 3600) // 60,
982
seconds=td.seconds % 60))
984
class PropertySetter(Base):
985
"Abstract class for Actions for setting one client property"
987
def run_on_one_client(self, client, properties=None):
988
"""Set the Client's D-Bus property"""
989
self.bus.set_client_property(client, self.propname,
994
raise NotImplementedError()
996
class Enable(PropertySetter):
1000
class Disable(PropertySetter):
1001
propname = "Enabled"
1002
value_to_set = False
1004
class BumpTimeout(PropertySetter):
1005
propname = "LastCheckedOK"
1008
class StartChecker(PropertySetter):
1009
propname = "CheckerRunning"
1012
class StopChecker(PropertySetter):
1013
propname = "CheckerRunning"
1014
value_to_set = False
1016
class ApproveByDefault(PropertySetter):
1017
propname = "ApprovedByDefault"
1020
class DenyByDefault(PropertySetter):
1021
propname = "ApprovedByDefault"
1022
value_to_set = False
1024
class PropertySetterValue(PropertySetter):
1025
"""Abstract class for PropertySetter recieving a value as
1026
constructor argument instead of a class attribute."""
1028
def __init__(self, value):
1029
self.value_to_set = value
1032
def argparse(cls, argtype):
1034
return cls(argtype(arg))
1037
class SetChecker(PropertySetterValue):
1038
propname = "Checker"
1040
class SetHost(PropertySetterValue):
1043
class SetSecret(PropertySetterValue):
1047
def value_to_set(self):
1050
@value_to_set.setter
1051
def value_to_set(self, value):
1052
"""When setting, read data from supplied file object"""
1053
self._vts = value.read()
1056
class PropertySetterValueMilliseconds(PropertySetterValue):
1057
"""Abstract class for PropertySetterValue taking a value
1058
argument as a datetime.timedelta() but should store it as
1062
def value_to_set(self):
1065
@value_to_set.setter
1066
def value_to_set(self, value):
1067
"When setting, convert value from a datetime.timedelta"
1068
self._vts = int(round(value.total_seconds() * 1000))
1070
class SetTimeout(PropertySetterValueMilliseconds):
1071
propname = "Timeout"
1073
class SetExtendedTimeout(PropertySetterValueMilliseconds):
1074
propname = "ExtendedTimeout"
1076
class SetInterval(PropertySetterValueMilliseconds):
1077
propname = "Interval"
1079
class SetApprovalDelay(PropertySetterValueMilliseconds):
1080
propname = "ApprovalDelay"
1082
class SetApprovalDuration(PropertySetterValueMilliseconds):
1083
propname = "ApprovalDuration"
734
class Test_milliseconds_to_string(unittest.TestCase):
736
self.assertEqual(milliseconds_to_string(93785000),
738
def test_no_days(self):
739
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
740
def test_all_zero(self):
741
self.assertEqual(milliseconds_to_string(0), "00:00:00")
742
def test_no_fractional_seconds(self):
743
self.assertEqual(milliseconds_to_string(400), "00:00:00")
744
self.assertEqual(milliseconds_to_string(900), "00:00:00")
745
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
747
class Test_string_to_delta(unittest.TestCase):
748
def test_handles_basic_rfc3339(self):
749
self.assertEqual(string_to_delta("PT2H"),
750
datetime.timedelta(0, 7200))
1087
class TestCaseWithAssertLogs(unittest.TestCase):
1088
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
1090
if not hasattr(unittest.TestCase, "assertLogs"):
1091
@contextlib.contextmanager
1092
def assertLogs(self, logger, level=logging.INFO):
1093
capturing_handler = self.CapturingLevelHandler(level)
1094
old_level = logger.level
1095
old_propagate = logger.propagate
1096
logger.addHandler(capturing_handler)
1097
logger.setLevel(level)
1098
logger.propagate = False
1100
yield capturing_handler.watcher
1102
logger.propagate = old_propagate
1103
logger.removeHandler(capturing_handler)
1104
logger.setLevel(old_level)
1105
self.assertGreater(len(capturing_handler.watcher.records),
1108
class CapturingLevelHandler(logging.Handler):
1109
def __init__(self, level, *args, **kwargs):
1110
logging.Handler.__init__(self, *args, **kwargs)
1111
self.watcher = self.LoggingWatcher([], [])
1112
def emit(self, record):
1113
self.watcher.records.append(record)
1114
self.watcher.output.append(self.format(record))
1116
LoggingWatcher = collections.namedtuple("LoggingWatcher",
1122
"""Class for objects which exist only to be unique objects, since
1123
unittest.mock.sentinel only exists in Python 3.3"""
1126
class Test_string_to_delta(TestCaseWithAssertLogs):
1127
# Just test basic RFC 3339 functionality here, the doc string for
1128
# rfc3339_duration_to_delta() already has more comprehensive
1129
# tests, which are run by doctest.
1131
def test_rfc3339_zero_seconds(self):
1132
self.assertEqual(datetime.timedelta(),
1133
string_to_delta("PT0S"))
1135
def test_rfc3339_zero_days(self):
1136
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
1138
def test_rfc3339_one_second(self):
1139
self.assertEqual(datetime.timedelta(0, 1),
1140
string_to_delta("PT1S"))
1142
def test_rfc3339_two_hours(self):
1143
self.assertEqual(datetime.timedelta(0, 7200),
1144
string_to_delta("PT2H"))
751
1146
def test_falls_back_to_pre_1_6_1_with_warning(self):
752
# assertLogs only exists in Python 3.4
753
if hasattr(self, "assertLogs"):
754
with self.assertLogs(log, logging.WARNING):
755
value = string_to_delta("2h")
1147
with self.assertLogs(log, logging.WARNING):
1148
value = string_to_delta("2h")
1149
self.assertEqual(datetime.timedelta(0, 7200), value)
1152
class Test_check_option_syntax(unittest.TestCase):
1154
self.parser = argparse.ArgumentParser()
1155
add_command_line_options(self.parser)
1157
def test_actions_requires_client_or_all(self):
1158
for action, value in self.actions.items():
1159
args = self.actionargs(action, value)
1160
with self.assertParseError():
1161
self.parse_args(args)
1163
# This mostly corresponds to the definition from has_commands() in
1164
# check_option_syntax()
1168
"--bump-timeout": None,
1169
"--start-checker": None,
1170
"--stop-checker": None,
1171
"--is-enabled": None,
1174
"--timeout": "PT0S",
1175
"--extended-timeout": "PT0S",
1176
"--interval": "PT0S",
1177
"--approve-by-default": None,
1178
"--deny-by-default": None,
1179
"--approval-delay": "PT0S",
1180
"--approval-duration": "PT0S",
1181
"--host": "hostname",
1182
"--secret": "/dev/null",
1188
def actionargs(action, value, *args):
1189
if value is not None:
1190
return [action, value] + list(args)
757
class WarningFilter(logging.Filter):
758
"""Don't show, but record the presence of, warnings"""
759
def filter(self, record):
760
is_warning = record.levelno >= logging.WARNING
761
self.found = is_warning or getattr(self, "found",
763
return not is_warning
764
warning_filter = WarningFilter()
765
log.addFilter(warning_filter)
1192
return [action] + list(args)
1194
@contextlib.contextmanager
1195
def assertParseError(self):
1196
with self.assertRaises(SystemExit) as e:
1197
with self.redirect_stderr_to_devnull():
1199
# Exit code from argparse is guaranteed to be "2". Reference:
1200
# https://docs.python.org/3/library
1201
# /argparse.html#exiting-methods
1202
self.assertEqual(2, e.exception.code)
1204
def parse_args(self, args):
1205
options = self.parser.parse_args(args)
1206
check_option_syntax(self.parser, options)
1209
@contextlib.contextmanager
1210
def redirect_stderr_to_devnull():
1211
old_stderr = sys.stderr
1212
with contextlib.closing(open(os.devnull, "w")) as null:
767
value = string_to_delta("2h")
769
log.removeFilter(warning_filter)
770
self.assertTrue(getattr(warning_filter, "found", False))
771
self.assertEqual(value, datetime.timedelta(0, 7200))
774
class TestCmd(unittest.TestCase):
1217
sys.stderr = old_stderr
1219
def check_option_syntax(self, options):
1220
check_option_syntax(self.parser, options)
1222
def test_actions_all_conflicts_with_verbose(self):
1223
for action, value in self.actions.items():
1224
args = self.actionargs(action, value, "--all",
1226
with self.assertParseError():
1227
self.parse_args(args)
1229
def test_actions_with_client_conflicts_with_verbose(self):
1230
for action, value in self.actions.items():
1231
args = self.actionargs(action, value, "--verbose",
1233
with self.assertParseError():
1234
self.parse_args(args)
1236
def test_dump_json_conflicts_with_verbose(self):
1237
args = ["--dump-json", "--verbose"]
1238
with self.assertParseError():
1239
self.parse_args(args)
1241
def test_dump_json_conflicts_with_action(self):
1242
for action, value in self.actions.items():
1243
args = self.actionargs(action, value, "--dump-json")
1244
with self.assertParseError():
1245
self.parse_args(args)
1247
def test_all_can_not_be_alone(self):
1249
with self.assertParseError():
1250
self.parse_args(args)
1252
def test_all_is_ok_with_any_action(self):
1253
for action, value in self.actions.items():
1254
args = self.actionargs(action, value, "--all")
1255
self.parse_args(args)
1257
def test_any_action_is_ok_with_one_client(self):
1258
for action, value in self.actions.items():
1259
args = self.actionargs(action, value, "client")
1260
self.parse_args(args)
1262
def test_one_client_with_all_actions_except_is_enabled(self):
1263
for action, value in self.actions.items():
1264
if action == "--is-enabled":
1266
args = self.actionargs(action, value, "client")
1267
self.parse_args(args)
1269
def test_two_clients_with_all_actions_except_is_enabled(self):
1270
for action, value in self.actions.items():
1271
if action == "--is-enabled":
1273
args = self.actionargs(action, value, "client1",
1275
self.parse_args(args)
1277
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1278
for action, value in self.actions.items():
1279
if action == "--is-enabled":
1281
args = self.actionargs(action, value, "client1",
1283
self.parse_args(args)
1285
def test_is_enabled_fails_without_client(self):
1286
args = ["--is-enabled"]
1287
with self.assertParseError():
1288
self.parse_args(args)
1290
def test_is_enabled_fails_with_two_clients(self):
1291
args = ["--is-enabled", "client1", "client2"]
1292
with self.assertParseError():
1293
self.parse_args(args)
1295
def test_remove_can_only_be_combined_with_action_deny(self):
1296
for action, value in self.actions.items():
1297
if action in {"--remove", "--deny"}:
1299
args = self.actionargs(action, value, "--all",
1301
with self.assertParseError():
1302
self.parse_args(args)
1305
class Test_dbus_exceptions(unittest.TestCase):
1307
def test_dbus_ConnectFailed_is_Error(self):
1308
with self.assertRaises(dbus.Error):
1309
raise dbus.ConnectFailed()
1312
class Test_dbus_MandosBus(unittest.TestCase):
1314
class MockMandosBus(dbus.MandosBus):
1316
self._name = "se.recompile.Mandos"
1317
self._server_path = "/"
1318
self._server_interface = "se.recompile.Mandos"
1319
self._client_interface = "se.recompile.Mandos.Client"
1321
self.call_method_return = Unique()
1323
def call_method(self, methodname, busname, objectpath,
1325
self.calls.append((methodname, busname, objectpath,
1327
return self.call_method_return
1330
self.bus = self.MockMandosBus()
1332
def test_set_client_property(self):
1333
self.bus.set_client_property("objectpath", "key", "value")
1334
expected_call = ("Set", self.bus._name, "objectpath",
1335
"org.freedesktop.DBus.Properties",
1336
(self.bus._client_interface, "key", "value"))
1337
self.assertIn(expected_call, self.bus.calls)
1339
def test_call_client_method(self):
1340
ret = self.bus.call_client_method("objectpath", "methodname")
1341
self.assertIs(self.bus.call_method_return, ret)
1342
expected_call = ("methodname", self.bus._name, "objectpath",
1343
self.bus._client_interface, ())
1344
self.assertIn(expected_call, self.bus.calls)
1346
def test_call_client_method_with_args(self):
1347
args = (Unique(), Unique())
1348
ret = self.bus.call_client_method("objectpath", "methodname",
1350
self.assertIs(self.bus.call_method_return, ret)
1351
expected_call = ("methodname", self.bus._name, "objectpath",
1352
self.bus._client_interface,
1354
self.assertIn(expected_call, self.bus.calls)
1356
def test_get_clients_and_properties(self):
1359
self.bus._client_interface: {
1363
"irrelevant_interface": {
1364
"key": "othervalue",
1368
"other_objectpath": {
1369
"other_irrelevant_interface": {
1375
expected_clients_and_properties = {
1381
self.bus.call_method_return = managed_objects
1382
ret = self.bus.get_clients_and_properties()
1383
self.assertDictEqual(expected_clients_and_properties, ret)
1384
expected_call = ("GetManagedObjects", self.bus._name,
1385
self.bus._server_path,
1386
"org.freedesktop.DBus.ObjectManager", ())
1387
self.assertIn(expected_call, self.bus.calls)
1389
def test_call_server_method(self):
1390
ret = self.bus.call_server_method("methodname")
1391
self.assertIs(self.bus.call_method_return, ret)
1392
expected_call = ("methodname", self.bus._name,
1393
self.bus._server_path,
1394
self.bus._server_interface, ())
1395
self.assertIn(expected_call, self.bus.calls)
1397
def test_call_server_method_with_args(self):
1398
args = (Unique(), Unique())
1399
ret = self.bus.call_server_method("methodname", *args)
1400
self.assertIs(self.bus.call_method_return, ret)
1401
expected_call = ("methodname", self.bus._name,
1402
self.bus._server_path,
1403
self.bus._server_interface,
1405
self.assertIn(expected_call, self.bus.calls)
1408
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1410
def MockDBusPython_func(self, func):
1411
class mock_dbus_python:
1412
"""mock dbus-python module"""
1414
"""Pseudo-namespace"""
1415
class DBusException(Exception):
1419
def get_object(busname, objectpath):
1420
DBusObject = collections.namedtuple(
1421
"DBusObject", ("methodname", "Set"))
1422
def method(*args, **kwargs):
1423
self.assertEqual({"dbus_interface":
1427
def set_property(interface, key, value,
1428
dbus_interface=None):
1430
"org.freedesktop.DBus.Properties",
1432
self.assertEqual("Secret", key)
1433
return func(interface, key, value,
1434
dbus_interface=dbus_interface)
1435
return DBusObject(methodname=method,
1438
def __init__(self, value):
1439
self.value = bool(value)
1442
if sys.version_info.major == 2:
1443
__nonzero__ = __bool__
1444
class ObjectPath(str):
1446
class Dictionary(dict):
1448
class ByteArray(bytes):
1450
return mock_dbus_python
1452
def call_method(self, bus, methodname, busname, objectpath,
1454
with self.assertLogs(log, logging.DEBUG):
1455
return bus.call_method(methodname, busname, objectpath,
1458
def test_call_method_returns(self):
1459
expected_method_return = Unique()
1460
method_args = (Unique(), Unique())
1462
self.assertEqual(len(method_args), len(args))
1463
for marg, arg in zip(method_args, args):
1464
self.assertIs(marg, arg)
1465
return expected_method_return
1466
mock_dbus_python = self.MockDBusPython_func(func)
1467
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1468
ret = self.call_method(bus, "methodname", "busname",
1469
"objectpath", "interface",
1471
self.assertIs(ret, expected_method_return)
1473
def test_call_method_filters_bool_true(self):
1475
return method_return
1476
mock_dbus_python = self.MockDBusPython_func(func)
1477
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1478
method_return = mock_dbus_python.Boolean(True)
1479
ret = self.call_method(bus, "methodname", "busname",
1480
"objectpath", "interface")
1481
self.assertTrue(ret)
1482
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1484
def test_call_method_filters_bool_false(self):
1486
return method_return
1487
mock_dbus_python = self.MockDBusPython_func(func)
1488
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1489
method_return = mock_dbus_python.Boolean(False)
1490
ret = self.call_method(bus, "methodname", "busname",
1491
"objectpath", "interface")
1492
self.assertFalse(ret)
1493
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1495
def test_call_method_filters_objectpath(self):
1497
return method_return
1498
mock_dbus_python = self.MockDBusPython_func(func)
1499
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1500
method_return = mock_dbus_python.ObjectPath("objectpath")
1501
ret = self.call_method(bus, "methodname", "busname",
1502
"objectpath", "interface")
1503
self.assertEqual("objectpath", ret)
1504
self.assertIsNot("objectpath", ret)
1505
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1507
def test_call_method_filters_booleans_in_dict(self):
1509
return method_return
1510
mock_dbus_python = self.MockDBusPython_func(func)
1511
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1512
method_return = mock_dbus_python.Dictionary(
1513
{mock_dbus_python.Boolean(True):
1514
mock_dbus_python.Boolean(False),
1515
mock_dbus_python.Boolean(False):
1516
mock_dbus_python.Boolean(True)})
1517
ret = self.call_method(bus, "methodname", "busname",
1518
"objectpath", "interface")
1519
expected_method_return = {True: False,
1521
self.assertEqual(expected_method_return, ret)
1522
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1524
def test_call_method_filters_objectpaths_in_dict(self):
1526
return method_return
1527
mock_dbus_python = self.MockDBusPython_func(func)
1528
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1529
method_return = mock_dbus_python.Dictionary(
1530
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1531
mock_dbus_python.ObjectPath("objectpath_value_1"),
1532
mock_dbus_python.ObjectPath("objectpath_key_2"):
1533
mock_dbus_python.ObjectPath("objectpath_value_2")})
1534
ret = self.call_method(bus, "methodname", "busname",
1535
"objectpath", "interface")
1536
expected_method_return = {str(key): str(value)
1538
method_return.items()}
1539
self.assertEqual(expected_method_return, ret)
1540
self.assertIsInstance(ret, dict)
1541
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1543
def test_call_method_filters_dict_in_dict(self):
1545
return method_return
1546
mock_dbus_python = self.MockDBusPython_func(func)
1547
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1548
method_return = mock_dbus_python.Dictionary(
1549
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1550
"key12": "value12"}),
1551
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1552
"key22": "value22"})})
1553
ret = self.call_method(bus, "methodname", "busname",
1554
"objectpath", "interface")
1555
expected_method_return = {
1556
"key1": {"key11": "value11",
1557
"key12": "value12"},
1558
"key2": {"key21": "value21",
1559
"key22": "value22"},
1561
self.assertEqual(expected_method_return, ret)
1562
self.assertIsInstance(ret, dict)
1563
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1564
for key, value in ret.items():
1565
self.assertIsInstance(value, dict)
1566
self.assertEqual(expected_method_return[key], value)
1567
self.assertNotIsInstance(value,
1568
mock_dbus_python.Dictionary)
1570
def test_call_method_filters_dict_three_deep(self):
1572
return method_return
1573
mock_dbus_python = self.MockDBusPython_func(func)
1574
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1575
method_return = mock_dbus_python.Dictionary(
1577
mock_dbus_python.Dictionary(
1579
mock_dbus_python.Dictionary(
1581
mock_dbus_python.Boolean(True),
1585
ret = self.call_method(bus, "methodname", "busname",
1586
"objectpath", "interface")
1587
expected_method_return = {"key1": {"key2": {"key3": True}}}
1588
self.assertEqual(expected_method_return, ret)
1589
self.assertIsInstance(ret, dict)
1590
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1591
self.assertIsInstance(ret["key1"], dict)
1592
self.assertNotIsInstance(ret["key1"],
1593
mock_dbus_python.Dictionary)
1594
self.assertIsInstance(ret["key1"]["key2"], dict)
1595
self.assertNotIsInstance(ret["key1"]["key2"],
1596
mock_dbus_python.Dictionary)
1597
self.assertTrue(ret["key1"]["key2"]["key3"])
1598
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1599
mock_dbus_python.Boolean)
1601
def test_call_method_handles_exception(self):
1602
dbus_logger = logging.getLogger("dbus.proxies")
1605
dbus_logger.error("Test")
1606
raise mock_dbus_python.exceptions.DBusException()
1608
mock_dbus_python = self.MockDBusPython_func(func)
1609
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1611
class CountingHandler(logging.Handler):
1613
def emit(self, record):
1616
counting_handler = CountingHandler()
1618
dbus_logger.addHandler(counting_handler)
1621
with self.assertRaises(dbus.Error) as e:
1622
self.call_method(bus, "methodname", "busname",
1623
"objectpath", "interface")
1625
dbus_logger.removeFilter(counting_handler)
1627
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1629
# Make sure the dbus logger was suppressed
1630
self.assertEqual(0, counting_handler.count)
1632
def test_Set_Secret_sends_bytearray(self):
1634
def func(*args, **kwargs):
1635
ret[0] = (args, kwargs)
1636
mock_dbus_python = self.MockDBusPython_func(func)
1637
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1638
bus.set_client_property("objectpath", "Secret", "value")
1639
expected_call = (("se.recompile.Mandos.Client", "Secret",
1640
mock_dbus_python.ByteArray(b"value")),
1642
"org.freedesktop.DBus.Properties"})
1643
self.assertEqual(expected_call, ret[0])
1644
if sys.version_info.major == 2:
1645
self.assertIsInstance(ret[0][0][-1],
1646
mock_dbus_python.ByteArray)
1648
def test_get_object_converts_to_correct_exception(self):
1649
bus = dbus_python_adapter.SystemBus(
1650
self.fake_dbus_python_raises_exception_on_connect)
1651
with self.assertRaises(dbus.ConnectFailed):
1652
self.call_method(bus, "methodname", "busname",
1653
"objectpath", "interface")
1655
class fake_dbus_python_raises_exception_on_connect:
1656
"""fake dbus-python module"""
1658
"""Pseudo-namespace"""
1659
class DBusException(Exception):
1664
def get_object(busname, objectpath):
1665
raise cls.exceptions.DBusException()
1666
Bus = collections.namedtuple("Bus", ["get_object"])
1667
return Bus(get_object=get_object)
1670
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1671
class mock_dbus_python:
1672
"""mock dbus-python modules"""
1675
def get_object(busname, objectpath):
1679
self.bus = dbus_python_adapter.CachingBus(
1680
self.mock_dbus_python)
1682
def test_returns_distinct_objectpaths(self):
1683
obj1 = self.bus.get_object("busname", "objectpath1")
1684
self.assertIsInstance(obj1, Unique)
1685
obj2 = self.bus.get_object("busname", "objectpath2")
1686
self.assertIsInstance(obj2, Unique)
1687
self.assertIsNot(obj1, obj2)
1689
def test_returns_distinct_busnames(self):
1690
obj1 = self.bus.get_object("busname1", "objectpath")
1691
self.assertIsInstance(obj1, Unique)
1692
obj2 = self.bus.get_object("busname2", "objectpath")
1693
self.assertIsInstance(obj2, Unique)
1694
self.assertIsNot(obj1, obj2)
1696
def test_returns_distinct_both(self):
1697
obj1 = self.bus.get_object("busname1", "objectpath")
1698
self.assertIsInstance(obj1, Unique)
1699
obj2 = self.bus.get_object("busname2", "objectpath")
1700
self.assertIsInstance(obj2, Unique)
1701
self.assertIsNot(obj1, obj2)
1703
def test_returns_same(self):
1704
obj1 = self.bus.get_object("busname", "objectpath")
1705
self.assertIsInstance(obj1, Unique)
1706
obj2 = self.bus.get_object("busname", "objectpath")
1707
self.assertIsInstance(obj2, Unique)
1708
self.assertIs(obj1, obj2)
1710
def test_returns_same_old(self):
1711
obj1 = self.bus.get_object("busname1", "objectpath1")
1712
self.assertIsInstance(obj1, Unique)
1713
obj2 = self.bus.get_object("busname2", "objectpath2")
1714
self.assertIsInstance(obj2, Unique)
1715
obj1b = self.bus.get_object("busname1", "objectpath1")
1716
self.assertIsInstance(obj1b, Unique)
1717
self.assertIsNot(obj1, obj2)
1718
self.assertIsNot(obj2, obj1b)
1719
self.assertIs(obj1, obj1b)
1722
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1724
def Stub_pydbus_func(self, func):
1726
"""stub pydbus module"""
1729
def get(busname, objectpath):
1730
DBusObject = collections.namedtuple(
1731
"DBusObject", ("methodname",))
1732
return {"interface":
1733
DBusObject(methodname=func)}
1736
def call_method(self, bus, methodname, busname, objectpath,
1738
with self.assertLogs(log, logging.DEBUG):
1739
return bus.call_method(methodname, busname, objectpath,
1742
def test_call_method_returns(self):
1743
expected_method_return = Unique()
1744
method_args = (Unique(), Unique())
1746
self.assertEqual(len(method_args), len(args))
1747
for marg, arg in zip(method_args, args):
1748
self.assertIs(marg, arg)
1749
return expected_method_return
1750
stub_pydbus = self.Stub_pydbus_func(func)
1751
bus = pydbus_adapter.SystemBus(stub_pydbus)
1752
ret = self.call_method(bus, "methodname", "busname",
1753
"objectpath", "interface",
1755
self.assertIs(ret, expected_method_return)
1757
def test_call_method_handles_exception(self):
1759
raise gi.repository.GLib.Error()
1761
stub_pydbus = self.Stub_pydbus_func(func)
1762
bus = pydbus_adapter.SystemBus(stub_pydbus)
1764
with self.assertRaises(dbus.Error) as e:
1765
self.call_method(bus, "methodname", "busname",
1766
"objectpath", "interface")
1768
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1770
def test_get_converts_to_correct_exception(self):
1771
bus = pydbus_adapter.SystemBus(
1772
self.fake_pydbus_raises_exception_on_connect)
1773
with self.assertRaises(dbus.ConnectFailed):
1774
self.call_method(bus, "methodname", "busname",
1775
"objectpath", "interface")
1777
class fake_pydbus_raises_exception_on_connect:
1778
"""fake dbus-python module"""
1781
def get(busname, objectpath):
1782
raise gi.repository.GLib.Error()
1783
Bus = collections.namedtuple("Bus", ["get"])
1786
def test_set_property_uses_setattr(self):
1793
def get(busname, objectpath):
1794
return {"interface": obj}
1795
bus = pydbus_adapter.SystemBus(pydbus_spy)
1797
bus.set_property("busname", "objectpath", "interface", "key",
1799
self.assertIs(value, obj.key)
1801
def test_get_suppresses_xml_deprecation_warning(self):
1802
if sys.version_info.major >= 3:
1804
class stub_pydbus_get:
1807
def get(busname, objectpath):
1808
warnings.warn_explicit(
1809
"deprecated", DeprecationWarning,
1810
"xml.etree.ElementTree", 0)
1811
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1812
with warnings.catch_warnings(record=True) as w:
1813
warnings.simplefilter("always")
1814
bus.get("busname", "objectpath")
1815
self.assertEqual(0, len(w))
1818
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1820
"""stub pydbus module"""
1823
def get(busname, objectpath):
1827
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1829
def test_returns_distinct_objectpaths(self):
1830
obj1 = self.bus.get("busname", "objectpath1")
1831
self.assertIsInstance(obj1, Unique)
1832
obj2 = self.bus.get("busname", "objectpath2")
1833
self.assertIsInstance(obj2, Unique)
1834
self.assertIsNot(obj1, obj2)
1836
def test_returns_distinct_busnames(self):
1837
obj1 = self.bus.get("busname1", "objectpath")
1838
self.assertIsInstance(obj1, Unique)
1839
obj2 = self.bus.get("busname2", "objectpath")
1840
self.assertIsInstance(obj2, Unique)
1841
self.assertIsNot(obj1, obj2)
1843
def test_returns_distinct_both(self):
1844
obj1 = self.bus.get("busname1", "objectpath")
1845
self.assertIsInstance(obj1, Unique)
1846
obj2 = self.bus.get("busname2", "objectpath")
1847
self.assertIsInstance(obj2, Unique)
1848
self.assertIsNot(obj1, obj2)
1850
def test_returns_same(self):
1851
obj1 = self.bus.get("busname", "objectpath")
1852
self.assertIsInstance(obj1, Unique)
1853
obj2 = self.bus.get("busname", "objectpath")
1854
self.assertIsInstance(obj2, Unique)
1855
self.assertIs(obj1, obj2)
1857
def test_returns_same_old(self):
1858
obj1 = self.bus.get("busname1", "objectpath1")
1859
self.assertIsInstance(obj1, Unique)
1860
obj2 = self.bus.get("busname2", "objectpath2")
1861
self.assertIsInstance(obj2, Unique)
1862
obj1b = self.bus.get("busname1", "objectpath1")
1863
self.assertIsInstance(obj1b, Unique)
1864
self.assertIsNot(obj1, obj2)
1865
self.assertIsNot(obj2, obj1b)
1866
self.assertIs(obj1, obj1b)
1869
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1873
class ObjectPath(str):
1875
class DBusError(Exception):
1878
def fake_ravel_func(self, func):
1882
class DBusInterfaceProxy:
1884
def methodname(*args):
1885
return [func(*args)]
1888
def get_interface(interface):
1889
if interface == "interface":
1890
return DBusInterfaceProxy()
1891
return {"busname": {"objectpath": DBusObject()}}
1894
def call_method(self, bus, methodname, busname, objectpath,
1896
with self.assertLogs(log, logging.DEBUG):
1897
return bus.call_method(methodname, busname, objectpath,
1900
def test_call_method_returns(self):
1901
expected_method_return = Unique()
1902
method_args = (Unique(), Unique())
1904
self.assertEqual(len(method_args), len(args))
1905
for marg, arg in zip(method_args, args):
1906
self.assertIs(marg, arg)
1907
return expected_method_return
1908
fake_ravel = self.fake_ravel_func(func)
1909
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1910
ret = self.call_method(bus, "methodname", "busname",
1911
"objectpath", "interface",
1913
self.assertIs(ret, expected_method_return)
1915
def test_call_method_filters_objectpath(self):
1917
return method_return
1918
fake_ravel = self.fake_ravel_func(func)
1919
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1920
method_return = (self.dummy_dbussy.DBUS
1921
.ObjectPath("objectpath"))
1922
ret = self.call_method(bus, "methodname", "busname",
1923
"objectpath", "interface")
1924
self.assertEqual("objectpath", ret)
1925
self.assertNotIsInstance(ret,
1926
self.dummy_dbussy.DBUS.ObjectPath)
1928
def test_call_method_filters_objectpaths_in_dict(self):
1929
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1931
return method_return
1932
fake_ravel = self.fake_ravel_func(func)
1933
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1935
ObjectPath("objectpath_key_1"):
1936
ObjectPath("objectpath_value_1"),
1937
ObjectPath("objectpath_key_2"):
1938
ObjectPath("objectpath_value_2"),
1940
ret = self.call_method(bus, "methodname", "busname",
1941
"objectpath", "interface")
1942
expected_method_return = {str(key): str(value)
1944
method_return.items()}
1945
for key, value in ret.items():
1946
self.assertNotIsInstance(key, ObjectPath)
1947
self.assertNotIsInstance(value, ObjectPath)
1948
self.assertEqual(expected_method_return, ret)
1949
self.assertIsInstance(ret, dict)
1951
def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1952
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1954
return method_return
1955
fake_ravel = self.fake_ravel_func(func)
1956
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1958
ObjectPath("key1"): {
1959
ObjectPath("key11"): ObjectPath("value11"),
1960
ObjectPath("key12"): ObjectPath("value12"),
1962
ObjectPath("key2"): {
1963
ObjectPath("key21"): ObjectPath("value21"),
1964
ObjectPath("key22"): ObjectPath("value22"),
1967
ret = self.call_method(bus, "methodname", "busname",
1968
"objectpath", "interface")
1969
expected_method_return = {
1970
"key1": {"key11": "value11",
1971
"key12": "value12"},
1972
"key2": {"key21": "value21",
1973
"key22": "value22"},
1975
self.assertEqual(expected_method_return, ret)
1976
for key, value in ret.items():
1977
self.assertIsInstance(value, dict)
1978
self.assertEqual(expected_method_return[key], value)
1979
self.assertNotIsInstance(key, ObjectPath)
1980
for inner_key, inner_value in value.items():
1981
self.assertIsInstance(value, dict)
1983
expected_method_return[key][inner_key],
1985
self.assertNotIsInstance(key, ObjectPath)
1987
def test_call_method_filters_objectpaths_in_dict_three_deep(self):
1988
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1990
return method_return
1991
fake_ravel = self.fake_ravel_func(func)
1992
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1994
ObjectPath("key1"): {
1995
ObjectPath("key2"): {
1996
ObjectPath("key3"): ObjectPath("value"),
2000
ret = self.call_method(bus, "methodname", "busname",
2001
"objectpath", "interface")
2002
expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2003
self.assertEqual(expected_method_return, ret)
2004
self.assertIsInstance(ret, dict)
2005
self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2006
self.assertIsInstance(ret["key1"], dict)
2007
self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2009
self.assertIsInstance(ret["key1"]["key2"], dict)
2010
self.assertNotIsInstance(
2011
next(iter(ret["key1"]["key2"].keys())),
2013
self.assertEqual("value", ret["key1"]["key2"]["key3"])
2014
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2015
self.dummy_dbussy.DBUS.ObjectPath)
2017
def test_call_method_handles_exception(self):
2019
raise self.dummy_dbussy.DBusError()
2021
fake_ravel = self.fake_ravel_func(func)
2022
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2024
with self.assertRaises(dbus.Error) as e:
2025
self.call_method(bus, "methodname", "busname",
2026
"objectpath", "interface")
2028
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2030
def test_get_object_converts_to_correct_exception(self):
2031
class fake_ravel_raises_exception_on_connect:
2036
def __getitem__(key):
2037
if key == "objectpath":
2038
raise self.dummy_dbussy.DBusError()
2039
raise Exception(key)
2040
return {"busname": Bus()}
2042
raise self.dummy_dbussy.DBusError()
2043
bus = dbussy_adapter.SystemBus(
2045
fake_ravel_raises_exception_on_connect)
2046
with self.assertRaises(dbus.ConnectFailed):
2047
self.call_method(bus, "methodname", "busname",
2048
"objectpath", "interface")
2051
class Test_commands_from_options(unittest.TestCase):
2054
self.parser = argparse.ArgumentParser()
2055
add_command_line_options(self.parser)
2057
def test_is_enabled(self):
2058
self.assert_command_from_args(["--is-enabled", "client"],
2061
def assert_command_from_args(self, args, command_cls, length=1,
2062
clients=None, **cmd_attrs):
2063
"""Assert that parsing ARGS should result in an instance of
2064
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
2065
options = self.parser.parse_args(args)
2066
check_option_syntax(self.parser, options)
2067
commands = commands_from_options(options)
2068
self.assertEqual(length, len(commands))
2069
for command in commands:
2070
if isinstance(command, command_cls):
2073
self.assertIsInstance(command, command_cls)
2074
if clients is not None:
2075
self.assertEqual(clients, options.client)
2076
for key, value in cmd_attrs.items():
2077
self.assertEqual(value, getattr(command, key))
2079
def assert_commands_from_args(self, args, commands, clients=None):
2080
for cmd in commands:
2081
self.assert_command_from_args(args, cmd,
2082
length=len(commands),
2085
def test_is_enabled_short(self):
2086
self.assert_command_from_args(["-V", "client"],
2089
def test_approve(self):
2090
self.assert_command_from_args(["--approve", "client"],
2093
def test_approve_short(self):
2094
self.assert_command_from_args(["-A", "client"],
2097
def test_deny(self):
2098
self.assert_command_from_args(["--deny", "client"],
2101
def test_deny_short(self):
2102
self.assert_command_from_args(["-D", "client"], command.Deny)
2104
def test_remove(self):
2105
self.assert_command_from_args(["--remove", "client"],
2108
def test_deny_before_remove(self):
2109
options = self.parser.parse_args(["--deny", "--remove",
2111
check_option_syntax(self.parser, options)
2112
commands = commands_from_options(options)
2113
self.assertEqual(2, len(commands))
2114
self.assertIsInstance(commands[0], command.Deny)
2115
self.assertIsInstance(commands[1], command.Remove)
2117
def test_deny_before_remove_reversed(self):
2118
options = self.parser.parse_args(["--remove", "--deny",
2120
check_option_syntax(self.parser, options)
2121
commands = commands_from_options(options)
2122
self.assertEqual(2, len(commands))
2123
self.assertIsInstance(commands[0], command.Deny)
2124
self.assertIsInstance(commands[1], command.Remove)
2126
def test_remove_short(self):
2127
self.assert_command_from_args(["-r", "client"],
2130
def test_dump_json(self):
2131
self.assert_command_from_args(["--dump-json"],
2134
def test_enable(self):
2135
self.assert_command_from_args(["--enable", "client"],
2138
def test_enable_short(self):
2139
self.assert_command_from_args(["-e", "client"],
2142
def test_disable(self):
2143
self.assert_command_from_args(["--disable", "client"],
2146
def test_disable_short(self):
2147
self.assert_command_from_args(["-d", "client"],
2150
def test_bump_timeout(self):
2151
self.assert_command_from_args(["--bump-timeout", "client"],
2152
command.BumpTimeout)
2154
def test_bump_timeout_short(self):
2155
self.assert_command_from_args(["-b", "client"],
2156
command.BumpTimeout)
2158
def test_start_checker(self):
2159
self.assert_command_from_args(["--start-checker", "client"],
2160
command.StartChecker)
2162
def test_stop_checker(self):
2163
self.assert_command_from_args(["--stop-checker", "client"],
2164
command.StopChecker)
2166
def test_approve_by_default(self):
2167
self.assert_command_from_args(["--approve-by-default",
2169
command.ApproveByDefault)
2171
def test_deny_by_default(self):
2172
self.assert_command_from_args(["--deny-by-default", "client"],
2173
command.DenyByDefault)
2175
def test_checker(self):
2176
self.assert_command_from_args(["--checker", ":", "client"],
2180
def test_checker_empty(self):
2181
self.assert_command_from_args(["--checker", "", "client"],
2185
def test_checker_short(self):
2186
self.assert_command_from_args(["-c", ":", "client"],
2190
def test_host(self):
2191
self.assert_command_from_args(
2192
["--host", "client.example.org", "client"],
2193
command.SetHost, value_to_set="client.example.org")
2195
def test_host_short(self):
2196
self.assert_command_from_args(
2197
["-H", "client.example.org", "client"], command.SetHost,
2198
value_to_set="client.example.org")
2200
def test_secret_devnull(self):
2201
self.assert_command_from_args(["--secret", os.path.devnull,
2202
"client"], command.SetSecret,
2205
def test_secret_tempfile(self):
2206
with tempfile.NamedTemporaryFile(mode="r+b") as f:
2207
value = b"secret\0xyzzy\nbar"
2210
self.assert_command_from_args(["--secret", f.name,
2215
def test_secret_devnull_short(self):
2216
self.assert_command_from_args(["-s", os.path.devnull,
2217
"client"], command.SetSecret,
2220
def test_secret_tempfile_short(self):
2221
with tempfile.NamedTemporaryFile(mode="r+b") as f:
2222
value = b"secret\0xyzzy\nbar"
2225
self.assert_command_from_args(["-s", f.name, "client"],
2229
def test_timeout(self):
2230
self.assert_command_from_args(["--timeout", "PT5M", "client"],
2232
value_to_set=300000)
2234
def test_timeout_short(self):
2235
self.assert_command_from_args(["-t", "PT5M", "client"],
2237
value_to_set=300000)
2239
def test_extended_timeout(self):
2240
self.assert_command_from_args(["--extended-timeout", "PT15M",
2242
command.SetExtendedTimeout,
2243
value_to_set=900000)
2245
def test_interval(self):
2246
self.assert_command_from_args(["--interval", "PT2M",
2247
"client"], command.SetInterval,
2248
value_to_set=120000)
2250
def test_interval_short(self):
2251
self.assert_command_from_args(["-i", "PT2M", "client"],
2252
command.SetInterval,
2253
value_to_set=120000)
2255
def test_approval_delay(self):
2256
self.assert_command_from_args(["--approval-delay", "PT30S",
2258
command.SetApprovalDelay,
2261
def test_approval_duration(self):
2262
self.assert_command_from_args(["--approval-duration", "PT1S",
2264
command.SetApprovalDuration,
2267
def test_print_table(self):
2268
self.assert_command_from_args([], command.PrintTable,
2271
def test_print_table_verbose(self):
2272
self.assert_command_from_args(["--verbose"],
2276
def test_print_table_verbose_short(self):
2277
self.assert_command_from_args(["-v"], command.PrintTable,
2281
def test_manual_page_example_1(self):
2282
self.assert_command_from_args("",
2287
def test_manual_page_example_2(self):
2288
self.assert_command_from_args(
2289
"--verbose foo1.example.org foo2.example.org".split(),
2290
command.PrintTable, clients=["foo1.example.org",
2291
"foo2.example.org"],
2294
def test_manual_page_example_3(self):
2295
self.assert_command_from_args("--enable --all".split(),
2299
def test_manual_page_example_4(self):
2300
self.assert_commands_from_args(
2301
("--timeout=PT5M --interval=PT1M foo1.example.org"
2302
" foo2.example.org").split(),
2303
[command.SetTimeout, command.SetInterval],
2304
clients=["foo1.example.org", "foo2.example.org"])
2306
def test_manual_page_example_5(self):
2307
self.assert_command_from_args("--approve --all".split(),
2312
class TestCommand(unittest.TestCase):
775
2313
"""Abstract class for tests of command classes"""
778
class MockClient(object):
779
def __init__(self, name, **attributes):
780
self.__dbus_object_path__ = "objpath_{}".format(name)
781
self.attributes = attributes
782
self.attributes["Name"] = name
784
def Set(self, interface, property, value, dbus_interface):
785
testcase.assertEqual(interface, client_interface)
786
testcase.assertEqual(dbus_interface,
787
dbus.PROPERTIES_IFACE)
788
self.attributes[property] = value
789
self.calls.append(("Set", (interface, property, value,
791
def Get(self, interface, property, dbus_interface):
792
testcase.assertEqual(interface, client_interface)
793
testcase.assertEqual(dbus_interface,
794
dbus.PROPERTIES_IFACE)
795
self.calls.append(("Get", (interface, property,
797
return self.attributes[property]
798
def __getitem__(self, key):
799
return self.attributes[key]
800
def __setitem__(self, key, value):
801
self.attributes[key] = value
802
self.clients = collections.OrderedDict([
806
KeyID=("92ed150794387c03ce684574b1139a65"
807
"94a34f895daaaf09fd8ea90a27cddb12"),
809
Host="foo.example.org",
810
Enabled=dbus.Boolean(True),
812
LastCheckedOK="2019-02-03T00:00:00",
813
Created="2019-01-02T00:00:00",
815
Fingerprint=("778827225BA7DE539C5A"
816
"7CFA59CFF7CDBD9A5920"),
817
CheckerRunning=dbus.Boolean(False),
818
LastEnabled="2019-01-03T00:00:00",
819
ApprovalPending=dbus.Boolean(False),
820
ApprovedByDefault=dbus.Boolean(True),
821
LastApprovalRequest="",
823
ApprovalDuration=1000,
824
Checker="fping -q -- %(host)s",
825
ExtendedTimeout=900000,
826
Expires="2019-02-04T00:00:00",
827
LastCheckerStatus=0)),
831
KeyID=("0558568eedd67d622f5c83b35a115f79"
832
"6ab612cff5ad227247e46c2b020f441c"),
835
Enabled=dbus.Boolean(True),
837
LastCheckedOK="2019-02-04T00:00:00",
838
Created="2019-01-03T00:00:00",
840
Fingerprint=("3E393AEAEFB84C7E89E2"
841
"F547B3A107558FCA3A27"),
842
CheckerRunning=dbus.Boolean(True),
843
LastEnabled="2019-01-04T00:00:00",
844
ApprovalPending=dbus.Boolean(False),
845
ApprovedByDefault=dbus.Boolean(False),
846
LastApprovalRequest="2019-01-03T00:00:00",
848
ApprovalDuration=1000,
850
ExtendedTimeout=900000,
851
Expires="2019-02-05T00:00:00",
852
LastCheckerStatus=-2)),
855
class TestPrintTableCmd(TestCmd):
856
def test_normal(self):
857
output = PrintTableCmd().output(self.clients)
858
expected_output = """
859
Name Enabled Timeout Last Successful Check
860
foo Yes 00:05:00 2019-02-03T00:00:00
861
barbar Yes 00:05:00 2019-02-04T00:00:00
863
self.assertEqual(output, expected_output)
864
def test_verbose(self):
865
output = PrintTableCmd(verbose=True).output(self.clients)
866
expected_output = """
867
Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status
868
foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0
869
barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2
871
self.assertEqual(output, expected_output)
872
def test_one_client(self):
873
output = PrintTableCmd().output({"foo": self.clients["foo"]})
874
expected_output = """
875
Name Enabled Timeout Last Successful Check
876
foo Yes 00:05:00 2019-02-03T00:00:00
878
self.assertEqual(output, expected_output)
880
class TestDumpJSONCmd(TestCmd):
882
self.expected_json = {
2315
class FakeMandosBus(dbus.MandosBus):
2316
def __init__(self, testcase):
2317
self.client_properties = {
885
2319
"KeyID": ("92ed150794387c03ce684574b1139a65"
886
2320
"94a34f895daaaf09fd8ea90a27cddb12"),
2321
"Secret": b"secret",
887
2322
"Host": "foo.example.org",
888
2323
"Enabled": True,
889
2324
"Timeout": 300000,