427
def check_option_syntax(parser, options):
428
"""Apply additional restrictions on options, not expressible in
431
def has_commands(options, commands=None):
433
commands = (command.Enable,
436
command.StartChecker,
442
command.SetExtendedTimeout,
444
command.ApproveByDefault,
445
command.DenyByDefault,
446
command.SetApprovalDelay,
447
command.SetApprovalDuration,
452
return any(isinstance(cmd, commands)
453
for cmd in options.commands)
455
if has_commands(options) and not (options.client or options.all):
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, properties in clients.items():
282
self.run_on_one_client(client, properties)
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, properties):
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):
326
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
327
keywords = default_keywords
329
keywords = self.all_keywords
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, properties):
423
if self.is_enabled(client, properties):
426
def is_enabled(self, client, properties):
427
return bool(properties["Enabled"])
429
class RemoveCmd(Command):
430
def run_on_one_client(self, client, properties):
431
self.mandos.RemoveClient(client.__dbus_object_path__)
433
class ApproveCmd(Command):
434
def run_on_one_client(self, client, properties):
435
client.Approve(dbus.Boolean(True),
436
dbus_interface=client_interface)
438
class DenyCmd(Command):
439
def run_on_one_client(self, client, properties):
440
client.Approve(dbus.Boolean(False),
441
dbus_interface=client_interface)
443
class EnableCmd(PropertyCmd):
445
value_to_set = dbus.Boolean(True)
447
class DisableCmd(PropertyCmd):
449
value_to_set = dbus.Boolean(False)
451
class BumpTimeoutCmd(PropertyCmd):
452
property = "LastCheckedOK"
455
class StartCheckerCmd(PropertyCmd):
456
property = "CheckerRunning"
457
value_to_set = dbus.Boolean(True)
459
class StopCheckerCmd(PropertyCmd):
460
property = "CheckerRunning"
461
value_to_set = dbus.Boolean(False)
463
class ApproveByDefaultCmd(PropertyCmd):
464
property = "ApprovedByDefault"
465
value_to_set = dbus.Boolean(True)
467
class DenyByDefaultCmd(PropertyCmd):
468
property = "ApprovedByDefault"
469
value_to_set = dbus.Boolean(False)
471
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
474
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
477
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
480
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
483
class SetExtendedTimeoutCmd(PropertyCmd,
484
MillisecondsValueArgumentMixIn):
485
property = "ExtendedTimeout"
487
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
488
property = "Interval"
490
class SetApprovalDelayCmd(PropertyCmd,
491
MillisecondsValueArgumentMixIn):
492
property = "ApprovalDelay"
494
class SetApprovalDurationCmd(PropertyCmd,
495
MillisecondsValueArgumentMixIn):
496
property = "ApprovalDuration"
498
def has_actions(options):
499
return any((options.enable,
501
options.bump_timeout,
502
options.start_checker,
503
options.stop_checker,
506
options.checker is not None,
507
options.timeout is not None,
508
options.extended_timeout is not None,
509
options.interval is not None,
510
options.approved_by_default is not None,
511
options.approval_delay is not None,
512
options.approval_duration is not None,
513
options.host is not None,
514
options.secret is not None,
518
def add_command_line_options(parser):
519
parser.add_argument("--version", action="version",
520
version="%(prog)s {}".format(version),
521
help="show version number and exit")
522
parser.add_argument("-a", "--all", action="store_true",
523
help="Select all clients")
524
parser.add_argument("-v", "--verbose", action="store_true",
525
help="Print all fields")
526
parser.add_argument("-j", "--dump-json", action="store_true",
527
help="Dump client data in JSON format")
528
enable_disable = parser.add_mutually_exclusive_group()
529
enable_disable.add_argument("-e", "--enable", action="store_true",
530
help="Enable client")
531
enable_disable.add_argument("-d", "--disable",
533
help="disable client")
534
parser.add_argument("-b", "--bump-timeout", action="store_true",
535
help="Bump timeout for client")
536
start_stop_checker = parser.add_mutually_exclusive_group()
537
start_stop_checker.add_argument("--start-checker",
539
help="Start checker for client")
540
start_stop_checker.add_argument("--stop-checker",
542
help="Stop checker for client")
543
parser.add_argument("-V", "--is-enabled", action="store_true",
544
help="Check if client is enabled")
545
parser.add_argument("-r", "--remove", action="store_true",
546
help="Remove client")
547
parser.add_argument("-c", "--checker",
548
help="Set checker command for client")
549
parser.add_argument("-t", "--timeout",
550
help="Set timeout for client")
551
parser.add_argument("--extended-timeout",
552
help="Set extended timeout for client")
553
parser.add_argument("-i", "--interval",
554
help="Set checker interval for client")
555
approve_deny_default = parser.add_mutually_exclusive_group()
556
approve_deny_default.add_argument(
557
"--approve-by-default", action="store_true",
558
default=None, dest="approved_by_default",
559
help="Set client to be approved by default")
560
approve_deny_default.add_argument(
561
"--deny-by-default", action="store_false",
562
dest="approved_by_default",
563
help="Set client to be denied by default")
564
parser.add_argument("--approval-delay",
565
help="Set delay before client approve/deny")
566
parser.add_argument("--approval-duration",
567
help="Set duration of one client approval")
568
parser.add_argument("-H", "--host", help="Set host for client")
569
parser.add_argument("-s", "--secret",
570
type=argparse.FileType(mode="rb"),
571
help="Set password blob (file) for client")
572
approve_deny = parser.add_mutually_exclusive_group()
573
approve_deny.add_argument(
574
"-A", "--approve", action="store_true",
575
help="Approve any current client request")
576
approve_deny.add_argument("-D", "--deny", action="store_true",
577
help="Deny any current client request")
578
parser.add_argument("--check", action="store_true",
579
help="Run self-test")
580
parser.add_argument("client", nargs="*", help="Client name")
583
def commands_from_options(options):
587
if options.dump_json:
588
commands.append(DumpJSONCmd())
591
commands.append(EnableCmd())
594
commands.append(DisableCmd())
596
if options.bump_timeout:
597
commands.append(BumpTimeoutCmd())
599
if options.start_checker:
600
commands.append(StartCheckerCmd())
602
if options.stop_checker:
603
commands.append(StopCheckerCmd())
605
if options.is_enabled:
606
commands.append(IsEnabledCmd())
609
commands.append(RemoveCmd())
611
if options.checker is not None:
612
commands.append(SetCheckerCmd())
614
if options.timeout is not None:
615
commands.append(SetTimeoutCmd(options.timeout))
617
if options.extended_timeout:
619
SetExtendedTimeoutCmd(options.extended_timeout))
621
if options.interval is not None:
622
command.append(SetIntervalCmd(options.interval))
624
if options.approved_by_default is not None:
625
if options.approved_by_default:
626
command.append(ApproveByDefaultCmd())
628
command.append(DenyByDefaultCmd())
630
if options.approval_delay is not None:
631
command.append(SetApprovalDelayCmd(options.approval_delay))
633
if options.approval_duration is not None:
635
SetApprovalDurationCmd(options.approval_duration))
637
if options.host is not None:
638
command.append(SetHostCmd(options.host))
640
if options.secret is not None:
641
command.append(SetSecretCmd(options.secret))
644
commands.append(ApproveCmd())
647
commands.append(DenyCmd())
649
# If no command option has been given, show table of clients,
650
# optionally verbosely
652
commands.append(PrintTableCmd(verbose=options.verbose))
658
parser = argparse.ArgumentParser()
660
add_command_line_options(parser)
662
options = parser.parse_args()
664
if has_actions(options) and not (options.client or options.all):
456
665
parser.error("Options require clients names or --all.")
457
if options.verbose and has_commands(options):
666
if options.verbose and has_actions(options):
458
667
parser.error("--verbose can only be used alone.")
459
if (has_commands(options, (command.DumpJSON,))
460
and (options.verbose or len(options.commands) > 1)):
668
if options.dump_json and (options.verbose
669
or has_actions(options)):
461
670
parser.error("--dump-json can only be used alone.")
462
if options.all and not has_commands(options):
671
if options.all and not has_actions(options):
463
672
parser.error("--all requires an action.")
464
if (has_commands(options, (command.IsEnabled,))
465
and len(options.client) > 1):
673
if options.is_enabled and len(options.client) > 1:
466
674
parser.error("--is-enabled requires exactly one client")
467
if (len(options.commands) > 1
468
and has_commands(options, (command.Remove,))
469
and not has_commands(options, (command.Deny,))):
470
parser.error("--remove can only be combined with --deny")
477
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
478
def get_managed_objects(self, busname, objectpath):
479
return self.call_method("GetManagedObjects", busname,
481
self.object_manager_iface)
483
properties_iface = "org.freedesktop.DBus.Properties"
484
def set_property(self, busname, objectpath, interface, key,
486
self.call_method("Set", busname, objectpath,
487
self.properties_iface, interface, key,
491
class MandosBus(SystemBus):
492
busname_domain = "se.recompile"
493
busname = busname_domain + ".Mandos"
495
server_interface = busname_domain + ".Mandos"
496
client_interface = busname_domain + ".Mandos.Client"
499
def get_clients_and_properties(self):
500
managed_objects = self.get_managed_objects(
501
self.busname, self.server_path)
502
return {objpath: properties[self.client_interface]
503
for objpath, properties in managed_objects.items()
504
if self.client_interface in properties}
506
def set_client_property(self, objectpath, key, value):
507
return self.set_property(self.busname, objectpath,
508
self.client_interface, key,
511
def call_client_method(self, objectpath, method, *args):
512
return self.call_method(method, self.busname, objectpath,
513
self.client_interface, *args)
515
def call_server_method(self, method, *args):
516
return self.call_method(method, self.busname,
518
self.server_interface, *args)
520
class Error(Exception):
523
class ConnectFailed(Error):
527
class dbus_python_adapter:
529
class SystemBus(dbus.MandosBus):
530
"""Use dbus-python"""
532
def __init__(self, module=dbus_python):
533
self.dbus_python = module
534
self.bus = self.dbus_python.SystemBus()
536
@contextlib.contextmanager
537
def convert_exception(self, exception_class=dbus.Error):
540
except self.dbus_python.exceptions.DBusException as e:
541
# This does what "raise from" would do
542
exc = exception_class(*e.args)
546
def call_method(self, methodname, busname, objectpath,
548
proxy_object = self.get_object(busname, objectpath)
549
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
550
interface, methodname,
551
", ".join(repr(a) for a in args))
552
method = getattr(proxy_object, methodname)
553
with self.convert_exception():
554
with dbus_python_adapter.SilenceLogger(
556
value = method(*args, dbus_interface=interface)
557
return self.type_filter(value)
559
def get_object(self, busname, objectpath):
560
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
562
with self.convert_exception(dbus.ConnectFailed):
563
return self.bus.get_object(busname, objectpath)
565
def type_filter(self, value):
566
"""Convert the most bothersome types to Python types"""
567
if isinstance(value, self.dbus_python.Boolean):
569
if isinstance(value, self.dbus_python.ObjectPath):
571
# Also recurse into dictionaries
572
if isinstance(value, self.dbus_python.Dictionary):
573
return {self.type_filter(key):
574
self.type_filter(subval)
575
for key, subval in value.items()}
578
def set_client_property(self, objectpath, key, value):
580
if not isinstance(value, bytes):
581
value = value.encode("utf-8")
582
value = self.dbus_python.ByteArray(value)
583
return self.set_property(self.busname, objectpath,
584
self.client_interface, key,
588
"Simple context manager to silence a particular logger"
589
def __init__(self, loggername):
590
self.logger = logging.getLogger(loggername)
593
self.logger.addFilter(self.nullfilter)
595
class NullFilter(logging.Filter):
596
def filter(self, record):
599
nullfilter = NullFilter()
601
def __exit__(self, exc_type, exc_val, exc_tb):
602
self.logger.removeFilter(self.nullfilter)
605
class CachingBus(SystemBus):
606
"""A caching layer for dbus_python_adapter.SystemBus"""
607
def __init__(self, *args, **kwargs):
608
self.object_cache = {}
609
super(dbus_python_adapter.CachingBus,
610
self).__init__(*args, **kwargs)
611
def get_object(self, busname, objectpath):
613
return self.object_cache[(busname, objectpath)]
616
dbus_python_adapter.CachingBus,
617
self).get_object(busname, objectpath)
618
self.object_cache[(busname, objectpath)] = new_object
622
class pydbus_adapter:
623
class SystemBus(dbus.MandosBus):
624
def __init__(self, module=pydbus):
626
self.bus = self.pydbus.SystemBus()
628
@contextlib.contextmanager
629
def convert_exception(self, exception_class=dbus.Error):
632
except gi.repository.GLib.Error as e:
633
# This does what "raise from" would do
634
exc = exception_class(*e.args)
638
def call_method(self, methodname, busname, objectpath,
640
proxy_object = self.get(busname, objectpath)
641
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
642
interface, methodname,
643
", ".join(repr(a) for a in args))
644
method = getattr(proxy_object[interface], methodname)
645
with self.convert_exception():
648
def get(self, busname, objectpath):
649
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
651
with self.convert_exception(dbus.ConnectFailed):
652
if sys.version_info.major <= 2:
653
with warnings.catch_warnings():
654
warnings.filterwarnings(
655
"ignore", "", DeprecationWarning,
656
r"^xml\.etree\.ElementTree$")
657
return self.bus.get(busname, objectpath)
659
return self.bus.get(busname, objectpath)
661
def set_property(self, busname, objectpath, interface, key,
663
proxy_object = self.get(busname, objectpath)
664
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
665
objectpath, self.properties_iface, interface,
667
setattr(proxy_object[interface], key, value)
669
class CachingBus(SystemBus):
670
"""A caching layer for pydbus_adapter.SystemBus"""
671
def __init__(self, *args, **kwargs):
672
self.object_cache = {}
673
super(pydbus_adapter.CachingBus,
674
self).__init__(*args, **kwargs)
675
def get(self, busname, objectpath):
677
return self.object_cache[(busname, objectpath)]
679
new_object = (super(pydbus_adapter.CachingBus, self)
680
.get(busname, objectpath))
681
self.object_cache[(busname, objectpath)] = new_object
685
def commands_from_options(options):
687
commands = list(options.commands)
689
def find_cmd(cmd, commands):
691
for i, c in enumerate(commands):
692
if isinstance(c, cmd):
696
# If command.Remove is present, move any instances of command.Deny
697
# to occur ahead of command.Remove.
698
index_of_remove = find_cmd(command.Remove, commands)
699
before_remove = commands[:index_of_remove]
700
after_remove = commands[index_of_remove:]
702
for cmd in after_remove:
703
if isinstance(cmd, command.Deny):
704
before_remove.append(cmd)
706
cleaned_after.append(cmd)
707
if cleaned_after != after_remove:
708
commands = before_remove + cleaned_after
710
# If no command option has been given, show table of clients,
711
# optionally verbosely
713
commands.append(command.PrintTable(verbose=options.verbose))
719
"""A namespace for command classes"""
722
"""Abstract base class for commands"""
723
def run(self, clients, bus=None):
724
"""Normal commands should implement run_on_one_client(),
725
but commands which want to operate on all clients at the same time can
726
override this run() method instead.
729
for client, properties in clients.items():
730
self.run_on_one_client(client, properties)
733
class IsEnabled(Base):
734
def run(self, clients, bus=None):
735
properties = next(iter(clients.values()))
736
if properties["Enabled"]:
742
def run_on_one_client(self, client, properties):
743
self.bus.call_client_method(client, "Approve", True)
747
def run_on_one_client(self, client, properties):
748
self.bus.call_client_method(client, "Approve", False)
752
def run(self, clients, bus):
753
for clientpath in frozenset(clients.keys()):
754
bus.call_server_method("RemoveClient", clientpath)
758
"""Abstract class for commands outputting client details"""
759
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
760
"Created", "Interval", "Host", "KeyID",
761
"Fingerprint", "CheckerRunning",
762
"LastEnabled", "ApprovalPending",
763
"ApprovedByDefault", "LastApprovalRequest",
764
"ApprovalDelay", "ApprovalDuration",
765
"Checker", "ExtendedTimeout", "Expires",
769
class DumpJSON(Output):
770
def run(self, clients, bus=None):
771
data = {properties["Name"]:
772
{key: properties[key]
773
for key in self.all_keywords}
774
for properties in clients.values()}
775
print(json.dumps(data, indent=4, separators=(',', ': ')))
778
class PrintTable(Output):
779
def __init__(self, verbose=False):
780
self.verbose = verbose
782
def run(self, clients, bus=None):
783
default_keywords = ("Name", "Enabled", "Timeout",
785
keywords = default_keywords
787
keywords = self.all_keywords
788
print(self.TableOfClients(clients.values(), keywords))
790
class TableOfClients:
793
"Enabled": "Enabled",
794
"Timeout": "Timeout",
795
"LastCheckedOK": "Last Successful Check",
796
"LastApprovalRequest": "Last Approval Request",
797
"Created": "Created",
798
"Interval": "Interval",
800
"Fingerprint": "Fingerprint",
802
"CheckerRunning": "Check Is Running",
803
"LastEnabled": "Last Enabled",
804
"ApprovalPending": "Approval Is Pending",
805
"ApprovedByDefault": "Approved By Default",
806
"ApprovalDelay": "Approval Delay",
807
"ApprovalDuration": "Approval Duration",
808
"Checker": "Checker",
809
"ExtendedTimeout": "Extended Timeout",
810
"Expires": "Expires",
811
"LastCheckerStatus": "Last Checker Status",
814
def __init__(self, clients, keywords):
815
self.clients = clients
816
self.keywords = keywords
819
return "\n".join(self.rows())
821
if sys.version_info.major == 2:
822
__unicode__ = __str__
824
return str(self).encode(
825
locale.getpreferredencoding())
828
format_string = self.row_formatting_string()
829
rows = [self.header_line(format_string)]
830
rows.extend(self.client_line(client, format_string)
831
for client in self.clients)
834
def row_formatting_string(self):
835
"Format string used to format table rows"
836
return " ".join("{{{key}:{width}}}".format(
837
width=max(len(self.tableheaders[key]),
838
*(len(self.string_from_client(client,
840
for client in self.clients)),
842
for key in self.keywords)
844
def string_from_client(self, client, key):
845
return self.valuetostring(client[key], key)
848
def valuetostring(cls, value, keyword):
849
if isinstance(value, bool):
850
return "Yes" if value else "No"
851
if keyword in ("Timeout", "Interval", "ApprovalDelay",
852
"ApprovalDuration", "ExtendedTimeout"):
853
return cls.milliseconds_to_string(value)
856
def header_line(self, format_string):
857
return format_string.format(**self.tableheaders)
859
def client_line(self, client, format_string):
860
return format_string.format(
861
**{key: self.string_from_client(client, key)
862
for key in self.keywords})
865
def milliseconds_to_string(ms):
866
td = datetime.timedelta(0, 0, 0, ms)
867
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
868
.format(days="{}T".format(td.days)
870
hours=td.seconds // 3600,
871
minutes=(td.seconds % 3600) // 60,
872
seconds=td.seconds % 60))
875
class PropertySetter(Base):
876
"Abstract class for Actions for setting one client property"
878
def run_on_one_client(self, client, properties=None):
879
"""Set the Client's D-Bus property"""
880
self.bus.set_client_property(client, self.propname,
885
raise NotImplementedError()
888
class Enable(PropertySetter):
893
class Disable(PropertySetter):
898
class BumpTimeout(PropertySetter):
899
propname = "LastCheckedOK"
903
class StartChecker(PropertySetter):
904
propname = "CheckerRunning"
908
class StopChecker(PropertySetter):
909
propname = "CheckerRunning"
913
class ApproveByDefault(PropertySetter):
914
propname = "ApprovedByDefault"
918
class DenyByDefault(PropertySetter):
919
propname = "ApprovedByDefault"
923
class PropertySetterValue(PropertySetter):
924
"""Abstract class for PropertySetter recieving a value as
925
constructor argument instead of a class attribute."""
926
def __init__(self, value):
927
self.value_to_set = value
930
def argparse(cls, argtype):
932
return cls(argtype(arg))
935
class SetChecker(PropertySetterValue):
939
class SetHost(PropertySetterValue):
943
class SetSecret(PropertySetterValue):
947
def value_to_set(self):
951
def value_to_set(self, value):
952
"""When setting, read data from supplied file object"""
953
self._vts = value.read()
957
class PropertySetterValueMilliseconds(PropertySetterValue):
958
"""Abstract class for PropertySetterValue taking a value
959
argument as a datetime.timedelta() but should store it as
963
def value_to_set(self):
967
def value_to_set(self, value):
968
"When setting, convert value from a datetime.timedelta"
969
self._vts = int(round(value.total_seconds() * 1000))
972
class SetTimeout(PropertySetterValueMilliseconds):
976
class SetExtendedTimeout(PropertySetterValueMilliseconds):
977
propname = "ExtendedTimeout"
980
class SetInterval(PropertySetterValueMilliseconds):
981
propname = "Interval"
984
class SetApprovalDelay(PropertySetterValueMilliseconds):
985
propname = "ApprovalDelay"
988
class SetApprovalDuration(PropertySetterValueMilliseconds):
989
propname = "ApprovalDuration"
676
clientnames = options.client
679
bus = dbus.SystemBus()
680
mandos_dbus_objc = bus.get_object(busname, server_path)
681
except dbus.exceptions.DBusException:
682
log.critical("Could not connect to Mandos server")
685
mandos_serv = dbus.Interface(mandos_dbus_objc,
686
dbus_interface=server_interface)
687
mandos_serv_object_manager = dbus.Interface(
688
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
690
# Filter out log message from dbus module
691
dbus_logger = logging.getLogger("dbus.proxies")
692
class NullFilter(logging.Filter):
693
def filter(self, record):
695
dbus_filter = NullFilter()
697
dbus_logger.addFilter(dbus_filter)
698
mandos_clients = {path: ifs_and_props[client_interface]
699
for path, ifs_and_props in
700
mandos_serv_object_manager
701
.GetManagedObjects().items()
702
if client_interface in ifs_and_props}
703
except dbus.exceptions.DBusException as e:
704
log.critical("Failed to access Mandos server through D-Bus:"
708
# restore dbus logger
709
dbus_logger.removeFilter(dbus_filter)
711
# Compile dict of (clients: properties) to process
715
clients = {bus.get_object(busname, path): properties
716
for path, properties in mandos_clients.items()}
718
for name in clientnames:
719
for path, client in mandos_clients.items():
720
if client["Name"] == name:
721
client_objc = bus.get_object(busname, path)
722
clients[client_objc] = client
725
log.critical("Client not found on server: %r", name)
728
# Run all commands on clients
729
commands = commands_from_options(options)
730
for command in commands:
731
command.run(mandos_serv, clients)
993
class TestCaseWithAssertLogs(unittest.TestCase):
994
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
996
if not hasattr(unittest.TestCase, "assertLogs"):
997
@contextlib.contextmanager
998
def assertLogs(self, logger, level=logging.INFO):
999
capturing_handler = self.CapturingLevelHandler(level)
1000
old_level = logger.level
1001
old_propagate = logger.propagate
1002
logger.addHandler(capturing_handler)
1003
logger.setLevel(level)
1004
logger.propagate = False
1006
yield capturing_handler.watcher
1008
logger.propagate = old_propagate
1009
logger.removeHandler(capturing_handler)
1010
logger.setLevel(old_level)
1011
self.assertGreater(len(capturing_handler.watcher.records),
1014
class CapturingLevelHandler(logging.Handler):
1015
def __init__(self, level, *args, **kwargs):
1016
logging.Handler.__init__(self, *args, **kwargs)
1017
self.watcher = self.LoggingWatcher([], [])
1018
def emit(self, record):
1019
self.watcher.records.append(record)
1020
self.watcher.output.append(self.format(record))
1022
LoggingWatcher = collections.namedtuple("LoggingWatcher",
1028
"""Class for objects which exist only to be unique objects, since
1029
unittest.mock.sentinel only exists in Python 3.3"""
1032
class Test_string_to_delta(TestCaseWithAssertLogs):
1033
# Just test basic RFC 3339 functionality here, the doc string for
1034
# rfc3339_duration_to_delta() already has more comprehensive
1035
# tests, which are run by doctest.
1037
def test_rfc3339_zero_seconds(self):
1038
self.assertEqual(datetime.timedelta(),
1039
string_to_delta("PT0S"))
1041
def test_rfc3339_zero_days(self):
1042
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
1044
def test_rfc3339_one_second(self):
1045
self.assertEqual(datetime.timedelta(0, 1),
1046
string_to_delta("PT1S"))
1048
def test_rfc3339_two_hours(self):
1049
self.assertEqual(datetime.timedelta(0, 7200),
1050
string_to_delta("PT2H"))
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("PT0S"),
750
datetime.timedelta())
751
self.assertEqual(string_to_delta("P0D"),
752
datetime.timedelta())
753
self.assertEqual(string_to_delta("PT1S"),
754
datetime.timedelta(0, 1))
755
self.assertEqual(string_to_delta("PT2H"),
756
datetime.timedelta(0, 7200))
1052
757
def test_falls_back_to_pre_1_6_1_with_warning(self):
1053
with self.assertLogs(log, logging.WARNING):
1054
value = string_to_delta("2h")
1055
self.assertEqual(datetime.timedelta(0, 7200), value)
1058
class Test_check_option_syntax(unittest.TestCase):
1060
self.parser = argparse.ArgumentParser()
1061
add_command_line_options(self.parser)
1063
def test_actions_requires_client_or_all(self):
1064
for action, value in self.actions.items():
1065
args = self.actionargs(action, value)
1066
with self.assertParseError():
1067
self.parse_args(args)
1069
# This mostly corresponds to the definition from has_commands() in
1070
# check_option_syntax()
1074
"--bump-timeout": None,
1075
"--start-checker": None,
1076
"--stop-checker": None,
1077
"--is-enabled": None,
1080
"--timeout": "PT0S",
1081
"--extended-timeout": "PT0S",
1082
"--interval": "PT0S",
1083
"--approve-by-default": None,
1084
"--deny-by-default": None,
1085
"--approval-delay": "PT0S",
1086
"--approval-duration": "PT0S",
1087
"--host": "hostname",
1088
"--secret": "/dev/null",
1094
def actionargs(action, value, *args):
1095
if value is not None:
1096
return [action, value] + list(args)
758
# assertLogs only exists in Python 3.4
759
if hasattr(self, "assertLogs"):
760
with self.assertLogs(log, logging.WARNING):
761
value = string_to_delta("2h")
1098
return [action] + list(args)
1100
@contextlib.contextmanager
1101
def assertParseError(self):
1102
with self.assertRaises(SystemExit) as e:
1103
with self.redirect_stderr_to_devnull():
1105
# Exit code from argparse is guaranteed to be "2". Reference:
1106
# https://docs.python.org/3/library
1107
# /argparse.html#exiting-methods
1108
self.assertEqual(2, e.exception.code)
1110
def parse_args(self, args):
1111
options = self.parser.parse_args(args)
1112
check_option_syntax(self.parser, options)
1115
@contextlib.contextmanager
1116
def redirect_stderr_to_devnull():
1117
old_stderr = sys.stderr
1118
with contextlib.closing(open(os.devnull, "w")) as null:
763
class WarningFilter(logging.Filter):
764
"""Don't show, but record the presence of, warnings"""
765
def filter(self, record):
766
is_warning = record.levelno >= logging.WARNING
767
self.found = is_warning or getattr(self, "found",
769
return not is_warning
770
warning_filter = WarningFilter()
771
log.addFilter(warning_filter)
773
value = string_to_delta("2h")
1123
sys.stderr = old_stderr
1125
def check_option_syntax(self, options):
1126
check_option_syntax(self.parser, options)
1128
def test_actions_all_conflicts_with_verbose(self):
1129
for action, value in self.actions.items():
1130
args = self.actionargs(action, value, "--all",
1132
with self.assertParseError():
1133
self.parse_args(args)
1135
def test_actions_with_client_conflicts_with_verbose(self):
1136
for action, value in self.actions.items():
1137
args = self.actionargs(action, value, "--verbose",
1139
with self.assertParseError():
1140
self.parse_args(args)
1142
def test_dump_json_conflicts_with_verbose(self):
1143
args = ["--dump-json", "--verbose"]
1144
with self.assertParseError():
1145
self.parse_args(args)
1147
def test_dump_json_conflicts_with_action(self):
1148
for action, value in self.actions.items():
1149
args = self.actionargs(action, value, "--dump-json")
1150
with self.assertParseError():
1151
self.parse_args(args)
1153
def test_all_can_not_be_alone(self):
1155
with self.assertParseError():
1156
self.parse_args(args)
1158
def test_all_is_ok_with_any_action(self):
1159
for action, value in self.actions.items():
1160
args = self.actionargs(action, value, "--all")
1161
self.parse_args(args)
1163
def test_any_action_is_ok_with_one_client(self):
1164
for action, value in self.actions.items():
1165
args = self.actionargs(action, value, "client")
1166
self.parse_args(args)
1168
def test_one_client_with_all_actions_except_is_enabled(self):
1169
for action, value in self.actions.items():
1170
if action == "--is-enabled":
1172
args = self.actionargs(action, value, "client")
1173
self.parse_args(args)
1175
def test_two_clients_with_all_actions_except_is_enabled(self):
1176
for action, value in self.actions.items():
1177
if action == "--is-enabled":
1179
args = self.actionargs(action, value, "client1",
1181
self.parse_args(args)
1183
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1184
for action, value in self.actions.items():
1185
if action == "--is-enabled":
1187
args = self.actionargs(action, value, "client1",
1189
self.parse_args(args)
1191
def test_is_enabled_fails_without_client(self):
1192
args = ["--is-enabled"]
1193
with self.assertParseError():
1194
self.parse_args(args)
1196
def test_is_enabled_fails_with_two_clients(self):
1197
args = ["--is-enabled", "client1", "client2"]
1198
with self.assertParseError():
1199
self.parse_args(args)
1201
def test_remove_can_only_be_combined_with_action_deny(self):
1202
for action, value in self.actions.items():
1203
if action in {"--remove", "--deny"}:
1205
args = self.actionargs(action, value, "--all",
1207
with self.assertParseError():
1208
self.parse_args(args)
1211
class Test_dbus_exceptions(unittest.TestCase):
1213
def test_dbus_ConnectFailed_is_Error(self):
1214
with self.assertRaises(dbus.Error):
1215
raise dbus.ConnectFailed()
1218
class Test_dbus_MandosBus(unittest.TestCase):
1220
class MockMandosBus(dbus.MandosBus):
1222
self._name = "se.recompile.Mandos"
1223
self._server_path = "/"
1224
self._server_interface = "se.recompile.Mandos"
1225
self._client_interface = "se.recompile.Mandos.Client"
1227
self.call_method_return = Unique()
1229
def call_method(self, methodname, busname, objectpath,
1231
self.calls.append((methodname, busname, objectpath,
1233
return self.call_method_return
1236
self.bus = self.MockMandosBus()
1238
def test_set_client_property(self):
1239
self.bus.set_client_property("objectpath", "key", "value")
1240
expected_call = ("Set", self.bus._name, "objectpath",
1241
"org.freedesktop.DBus.Properties",
1242
(self.bus._client_interface, "key", "value"))
1243
self.assertIn(expected_call, self.bus.calls)
1245
def test_call_client_method(self):
1246
ret = self.bus.call_client_method("objectpath", "methodname")
1247
self.assertIs(self.bus.call_method_return, ret)
1248
expected_call = ("methodname", self.bus._name, "objectpath",
1249
self.bus._client_interface, ())
1250
self.assertIn(expected_call, self.bus.calls)
1252
def test_call_client_method_with_args(self):
1253
args = (Unique(), Unique())
1254
ret = self.bus.call_client_method("objectpath", "methodname",
1256
self.assertIs(self.bus.call_method_return, ret)
1257
expected_call = ("methodname", self.bus._name, "objectpath",
1258
self.bus._client_interface,
1260
self.assertIn(expected_call, self.bus.calls)
1262
def test_get_clients_and_properties(self):
1265
self.bus._client_interface: {
1269
"irrelevant_interface": {
1270
"key": "othervalue",
1274
"other_objectpath": {
1275
"other_irrelevant_interface": {
1281
expected_clients_and_properties = {
1287
self.bus.call_method_return = managed_objects
1288
ret = self.bus.get_clients_and_properties()
1289
self.assertDictEqual(expected_clients_and_properties, ret)
1290
expected_call = ("GetManagedObjects", self.bus._name,
1291
self.bus._server_path,
1292
"org.freedesktop.DBus.ObjectManager", ())
1293
self.assertIn(expected_call, self.bus.calls)
1295
def test_call_server_method(self):
1296
ret = self.bus.call_server_method("methodname")
1297
self.assertIs(self.bus.call_method_return, ret)
1298
expected_call = ("methodname", self.bus._name,
1299
self.bus._server_path,
1300
self.bus._server_interface, ())
1301
self.assertIn(expected_call, self.bus.calls)
1303
def test_call_server_method_with_args(self):
1304
args = (Unique(), Unique())
1305
ret = self.bus.call_server_method("methodname", *args)
1306
self.assertIs(self.bus.call_method_return, ret)
1307
expected_call = ("methodname", self.bus._name,
1308
self.bus._server_path,
1309
self.bus._server_interface,
1311
self.assertIn(expected_call, self.bus.calls)
1314
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1316
def MockDBusPython_func(self, func):
1317
class mock_dbus_python:
1318
"""mock dbus-python module"""
1320
"""Pseudo-namespace"""
1321
class DBusException(Exception):
1325
def get_object(busname, objectpath):
1326
DBusObject = collections.namedtuple(
1327
"DBusObject", ("methodname", "Set"))
1328
def method(*args, **kwargs):
1329
self.assertEqual({"dbus_interface":
1333
def set_property(interface, key, value,
1334
dbus_interface=None):
1336
"org.freedesktop.DBus.Properties",
1338
self.assertEqual("Secret", key)
1339
return func(interface, key, value,
1340
dbus_interface=dbus_interface)
1341
return DBusObject(methodname=method,
1344
def __init__(self, value):
1345
self.value = bool(value)
1348
if sys.version_info.major == 2:
1349
__nonzero__ = __bool__
1350
class ObjectPath(str):
1352
class Dictionary(dict):
1354
class ByteArray(bytes):
1356
return mock_dbus_python
1358
def call_method(self, bus, methodname, busname, objectpath,
1360
with self.assertLogs(log, logging.DEBUG):
1361
return bus.call_method(methodname, busname, objectpath,
1364
def test_call_method_returns(self):
1365
expected_method_return = Unique()
1366
method_args = (Unique(), Unique())
1368
self.assertEqual(len(method_args), len(args))
1369
for marg, arg in zip(method_args, args):
1370
self.assertIs(marg, arg)
1371
return expected_method_return
1372
mock_dbus_python = self.MockDBusPython_func(func)
1373
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1374
ret = self.call_method(bus, "methodname", "busname",
1375
"objectpath", "interface",
1377
self.assertIs(ret, expected_method_return)
1379
def test_call_method_filters_bool_true(self):
1381
return method_return
1382
mock_dbus_python = self.MockDBusPython_func(func)
1383
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1384
method_return = mock_dbus_python.Boolean(True)
1385
ret = self.call_method(bus, "methodname", "busname",
1386
"objectpath", "interface")
1387
self.assertTrue(ret)
1388
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1390
def test_call_method_filters_bool_false(self):
1392
return method_return
1393
mock_dbus_python = self.MockDBusPython_func(func)
1394
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1395
method_return = mock_dbus_python.Boolean(False)
1396
ret = self.call_method(bus, "methodname", "busname",
1397
"objectpath", "interface")
1398
self.assertFalse(ret)
1399
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1401
def test_call_method_filters_objectpath(self):
1403
return method_return
1404
mock_dbus_python = self.MockDBusPython_func(func)
1405
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1406
method_return = mock_dbus_python.ObjectPath("objectpath")
1407
ret = self.call_method(bus, "methodname", "busname",
1408
"objectpath", "interface")
1409
self.assertEqual("objectpath", ret)
1410
self.assertIsNot("objectpath", ret)
1411
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1413
def test_call_method_filters_booleans_in_dict(self):
1415
return method_return
1416
mock_dbus_python = self.MockDBusPython_func(func)
1417
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1418
method_return = mock_dbus_python.Dictionary(
1419
{mock_dbus_python.Boolean(True):
1420
mock_dbus_python.Boolean(False),
1421
mock_dbus_python.Boolean(False):
1422
mock_dbus_python.Boolean(True)})
1423
ret = self.call_method(bus, "methodname", "busname",
1424
"objectpath", "interface")
1425
expected_method_return = {True: False,
1427
self.assertEqual(expected_method_return, ret)
1428
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1430
def test_call_method_filters_objectpaths_in_dict(self):
1432
return method_return
1433
mock_dbus_python = self.MockDBusPython_func(func)
1434
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1435
method_return = mock_dbus_python.Dictionary(
1436
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1437
mock_dbus_python.ObjectPath("objectpath_value_1"),
1438
mock_dbus_python.ObjectPath("objectpath_key_2"):
1439
mock_dbus_python.ObjectPath("objectpath_value_2")})
1440
ret = self.call_method(bus, "methodname", "busname",
1441
"objectpath", "interface")
1442
expected_method_return = {str(key): str(value)
1444
method_return.items()}
1445
self.assertEqual(expected_method_return, ret)
1446
self.assertIsInstance(ret, dict)
1447
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1449
def test_call_method_filters_dict_in_dict(self):
1451
return method_return
1452
mock_dbus_python = self.MockDBusPython_func(func)
1453
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1454
method_return = mock_dbus_python.Dictionary(
1455
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1456
"key12": "value12"}),
1457
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1458
"key22": "value22"})})
1459
ret = self.call_method(bus, "methodname", "busname",
1460
"objectpath", "interface")
1461
expected_method_return = {
1462
"key1": {"key11": "value11",
1463
"key12": "value12"},
1464
"key2": {"key21": "value21",
1465
"key22": "value22"},
1467
self.assertEqual(expected_method_return, ret)
1468
self.assertIsInstance(ret, dict)
1469
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1470
for key, value in ret.items():
1471
self.assertIsInstance(value, dict)
1472
self.assertEqual(expected_method_return[key], value)
1473
self.assertNotIsInstance(value,
1474
mock_dbus_python.Dictionary)
1476
def test_call_method_filters_dict_three_deep(self):
1478
return method_return
1479
mock_dbus_python = self.MockDBusPython_func(func)
1480
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1481
method_return = mock_dbus_python.Dictionary(
1483
mock_dbus_python.Dictionary(
1485
mock_dbus_python.Dictionary(
1487
mock_dbus_python.Boolean(True),
1491
ret = self.call_method(bus, "methodname", "busname",
1492
"objectpath", "interface")
1493
expected_method_return = {"key1": {"key2": {"key3": True}}}
1494
self.assertEqual(expected_method_return, ret)
1495
self.assertIsInstance(ret, dict)
1496
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1497
self.assertIsInstance(ret["key1"], dict)
1498
self.assertNotIsInstance(ret["key1"],
1499
mock_dbus_python.Dictionary)
1500
self.assertIsInstance(ret["key1"]["key2"], dict)
1501
self.assertNotIsInstance(ret["key1"]["key2"],
1502
mock_dbus_python.Dictionary)
1503
self.assertTrue(ret["key1"]["key2"]["key3"])
1504
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1505
mock_dbus_python.Boolean)
1507
def test_call_method_handles_exception(self):
1508
dbus_logger = logging.getLogger("dbus.proxies")
1511
dbus_logger.error("Test")
1512
raise mock_dbus_python.exceptions.DBusException()
1514
mock_dbus_python = self.MockDBusPython_func(func)
1515
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1517
class CountingHandler(logging.Handler):
1519
def emit(self, record):
1522
counting_handler = CountingHandler()
1524
dbus_logger.addHandler(counting_handler)
1527
with self.assertRaises(dbus.Error) as e:
1528
self.call_method(bus, "methodname", "busname",
1529
"objectpath", "interface")
1531
dbus_logger.removeFilter(counting_handler)
1533
self.assertNotIsInstance(e, dbus.ConnectFailed)
1535
# Make sure the dbus logger was suppressed
1536
self.assertEqual(0, counting_handler.count)
1538
def test_Set_Secret_sends_bytearray(self):
1540
def func(*args, **kwargs):
1541
ret[0] = (args, kwargs)
1542
mock_dbus_python = self.MockDBusPython_func(func)
1543
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1544
bus.set_client_property("objectpath", "Secret", "value")
1545
expected_call = (("se.recompile.Mandos.Client", "Secret",
1546
mock_dbus_python.ByteArray(b"value")),
1548
"org.freedesktop.DBus.Properties"})
1549
self.assertEqual(expected_call, ret[0])
1550
if sys.version_info.major == 2:
1551
self.assertIsInstance(ret[0][0][-1],
1552
mock_dbus_python.ByteArray)
1554
def test_get_object_converts_to_correct_exception(self):
1555
bus = dbus_python_adapter.SystemBus(
1556
self.fake_dbus_python_raises_exception_on_connect)
1557
with self.assertRaises(dbus.ConnectFailed):
1558
self.call_method(bus, "methodname", "busname",
1559
"objectpath", "interface")
1561
class fake_dbus_python_raises_exception_on_connect:
1562
"""fake dbus-python module"""
1564
"""Pseudo-namespace"""
1565
class DBusException(Exception):
1570
def get_object(busname, objectpath):
1571
raise cls.exceptions.DBusException()
1572
Bus = collections.namedtuple("Bus", ["get_object"])
1573
return Bus(get_object=get_object)
1576
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1577
class mock_dbus_python:
1578
"""mock dbus-python modules"""
1581
def get_object(busname, objectpath):
1585
self.bus = dbus_python_adapter.CachingBus(
1586
self.mock_dbus_python)
1588
def test_returns_distinct_objectpaths(self):
1589
obj1 = self.bus.get_object("busname", "objectpath1")
1590
self.assertIsInstance(obj1, Unique)
1591
obj2 = self.bus.get_object("busname", "objectpath2")
1592
self.assertIsInstance(obj2, Unique)
1593
self.assertIsNot(obj1, obj2)
1595
def test_returns_distinct_busnames(self):
1596
obj1 = self.bus.get_object("busname1", "objectpath")
1597
self.assertIsInstance(obj1, Unique)
1598
obj2 = self.bus.get_object("busname2", "objectpath")
1599
self.assertIsInstance(obj2, Unique)
1600
self.assertIsNot(obj1, obj2)
1602
def test_returns_distinct_both(self):
1603
obj1 = self.bus.get_object("busname1", "objectpath")
1604
self.assertIsInstance(obj1, Unique)
1605
obj2 = self.bus.get_object("busname2", "objectpath")
1606
self.assertIsInstance(obj2, Unique)
1607
self.assertIsNot(obj1, obj2)
1609
def test_returns_same(self):
1610
obj1 = self.bus.get_object("busname", "objectpath")
1611
self.assertIsInstance(obj1, Unique)
1612
obj2 = self.bus.get_object("busname", "objectpath")
1613
self.assertIsInstance(obj2, Unique)
1614
self.assertIs(obj1, obj2)
1616
def test_returns_same_old(self):
1617
obj1 = self.bus.get_object("busname1", "objectpath1")
1618
self.assertIsInstance(obj1, Unique)
1619
obj2 = self.bus.get_object("busname2", "objectpath2")
1620
self.assertIsInstance(obj2, Unique)
1621
obj1b = self.bus.get_object("busname1", "objectpath1")
1622
self.assertIsInstance(obj1b, Unique)
1623
self.assertIsNot(obj1, obj2)
1624
self.assertIsNot(obj2, obj1b)
1625
self.assertIs(obj1, obj1b)
1628
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1630
def Stub_pydbus_func(self, func):
1632
"""stub pydbus module"""
1635
def get(busname, objectpath):
1636
DBusObject = collections.namedtuple(
1637
"DBusObject", ("methodname",))
1638
return {"interface":
1639
DBusObject(methodname=func)}
1642
def call_method(self, bus, methodname, busname, objectpath,
1644
with self.assertLogs(log, logging.DEBUG):
1645
return bus.call_method(methodname, busname, objectpath,
1648
def test_call_method_returns(self):
1649
expected_method_return = Unique()
1650
method_args = (Unique(), Unique())
1652
self.assertEqual(len(method_args), len(args))
1653
for marg, arg in zip(method_args, args):
1654
self.assertIs(marg, arg)
1655
return expected_method_return
1656
stub_pydbus = self.Stub_pydbus_func(func)
1657
bus = pydbus_adapter.SystemBus(stub_pydbus)
1658
ret = self.call_method(bus, "methodname", "busname",
1659
"objectpath", "interface",
1661
self.assertIs(ret, expected_method_return)
1663
def test_call_method_handles_exception(self):
1664
dbus_logger = logging.getLogger("dbus.proxies")
1667
raise gi.repository.GLib.Error()
1669
stub_pydbus = self.Stub_pydbus_func(func)
1670
bus = pydbus_adapter.SystemBus(stub_pydbus)
1672
with self.assertRaises(dbus.Error) as e:
1673
self.call_method(bus, "methodname", "busname",
1674
"objectpath", "interface")
1676
self.assertNotIsInstance(e, dbus.ConnectFailed)
1678
def test_get_converts_to_correct_exception(self):
1679
bus = pydbus_adapter.SystemBus(
1680
self.fake_pydbus_raises_exception_on_connect)
1681
with self.assertRaises(dbus.ConnectFailed):
1682
self.call_method(bus, "methodname", "busname",
1683
"objectpath", "interface")
1685
class fake_pydbus_raises_exception_on_connect:
1686
"""fake dbus-python module"""
1689
def get(busname, objectpath):
1690
raise gi.repository.GLib.Error()
1691
Bus = collections.namedtuple("Bus", ["get"])
1694
def test_set_property_uses_setattr(self):
1701
def get(busname, objectpath):
1702
return {"interface": obj}
1703
bus = pydbus_adapter.SystemBus(pydbus_spy)
1705
bus.set_property("busname", "objectpath", "interface", "key",
1707
self.assertIs(value, obj.key)
1709
def test_get_suppresses_xml_deprecation_warning(self):
1710
if sys.version_info.major >= 3:
1712
class stub_pydbus_get:
1715
def get(busname, objectpath):
1716
warnings.warn_explicit(
1717
"deprecated", DeprecationWarning,
1718
"xml.etree.ElementTree", 0)
1719
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1720
with warnings.catch_warnings(record=True) as w:
1721
warnings.simplefilter("always")
1722
bus.get("busname", "objectpath")
1723
self.assertEqual(0, len(w))
1726
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1728
"""stub pydbus module"""
1731
def get(busname, objectpath):
1735
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1737
def test_returns_distinct_objectpaths(self):
1738
obj1 = self.bus.get("busname", "objectpath1")
1739
self.assertIsInstance(obj1, Unique)
1740
obj2 = self.bus.get("busname", "objectpath2")
1741
self.assertIsInstance(obj2, Unique)
1742
self.assertIsNot(obj1, obj2)
1744
def test_returns_distinct_busnames(self):
1745
obj1 = self.bus.get("busname1", "objectpath")
1746
self.assertIsInstance(obj1, Unique)
1747
obj2 = self.bus.get("busname2", "objectpath")
1748
self.assertIsInstance(obj2, Unique)
1749
self.assertIsNot(obj1, obj2)
1751
def test_returns_distinct_both(self):
1752
obj1 = self.bus.get("busname1", "objectpath")
1753
self.assertIsInstance(obj1, Unique)
1754
obj2 = self.bus.get("busname2", "objectpath")
1755
self.assertIsInstance(obj2, Unique)
1756
self.assertIsNot(obj1, obj2)
1758
def test_returns_same(self):
1759
obj1 = self.bus.get("busname", "objectpath")
1760
self.assertIsInstance(obj1, Unique)
1761
obj2 = self.bus.get("busname", "objectpath")
1762
self.assertIsInstance(obj2, Unique)
1763
self.assertIs(obj1, obj2)
1765
def test_returns_same_old(self):
1766
obj1 = self.bus.get("busname1", "objectpath1")
1767
self.assertIsInstance(obj1, Unique)
1768
obj2 = self.bus.get("busname2", "objectpath2")
1769
self.assertIsInstance(obj2, Unique)
1770
obj1b = self.bus.get("busname1", "objectpath1")
1771
self.assertIsInstance(obj1b, Unique)
1772
self.assertIsNot(obj1, obj2)
1773
self.assertIsNot(obj2, obj1b)
1774
self.assertIs(obj1, obj1b)
1777
class Test_commands_from_options(unittest.TestCase):
1780
self.parser = argparse.ArgumentParser()
1781
add_command_line_options(self.parser)
1783
def test_is_enabled(self):
1784
self.assert_command_from_args(["--is-enabled", "client"],
1787
def assert_command_from_args(self, args, command_cls, length=1,
1788
clients=None, **cmd_attrs):
1789
"""Assert that parsing ARGS should result in an instance of
1790
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1791
options = self.parser.parse_args(args)
1792
check_option_syntax(self.parser, options)
1793
commands = commands_from_options(options)
1794
self.assertEqual(length, len(commands))
1795
for command in commands:
1796
if isinstance(command, command_cls):
1799
self.assertIsInstance(command, command_cls)
1800
if clients is not None:
1801
self.assertEqual(clients, options.client)
1802
for key, value in cmd_attrs.items():
1803
self.assertEqual(value, getattr(command, key))
1805
def assert_commands_from_args(self, args, commands, clients=None):
1806
for cmd in commands:
1807
self.assert_command_from_args(args, cmd,
1808
length=len(commands),
1811
def test_is_enabled_short(self):
1812
self.assert_command_from_args(["-V", "client"],
1815
def test_approve(self):
1816
self.assert_command_from_args(["--approve", "client"],
1819
def test_approve_short(self):
1820
self.assert_command_from_args(["-A", "client"],
1823
def test_deny(self):
1824
self.assert_command_from_args(["--deny", "client"],
1827
def test_deny_short(self):
1828
self.assert_command_from_args(["-D", "client"], command.Deny)
1830
def test_remove(self):
1831
self.assert_command_from_args(["--remove", "client"],
1834
def test_deny_before_remove(self):
1835
options = self.parser.parse_args(["--deny", "--remove",
1837
check_option_syntax(self.parser, options)
1838
commands = commands_from_options(options)
1839
self.assertEqual(2, len(commands))
1840
self.assertIsInstance(commands[0], command.Deny)
1841
self.assertIsInstance(commands[1], command.Remove)
1843
def test_deny_before_remove_reversed(self):
1844
options = self.parser.parse_args(["--remove", "--deny",
1846
check_option_syntax(self.parser, options)
1847
commands = commands_from_options(options)
1848
self.assertEqual(2, len(commands))
1849
self.assertIsInstance(commands[0], command.Deny)
1850
self.assertIsInstance(commands[1], command.Remove)
1852
def test_remove_short(self):
1853
self.assert_command_from_args(["-r", "client"],
1856
def test_dump_json(self):
1857
self.assert_command_from_args(["--dump-json"],
1860
def test_enable(self):
1861
self.assert_command_from_args(["--enable", "client"],
1864
def test_enable_short(self):
1865
self.assert_command_from_args(["-e", "client"],
1868
def test_disable(self):
1869
self.assert_command_from_args(["--disable", "client"],
1872
def test_disable_short(self):
1873
self.assert_command_from_args(["-d", "client"],
1876
def test_bump_timeout(self):
1877
self.assert_command_from_args(["--bump-timeout", "client"],
1878
command.BumpTimeout)
1880
def test_bump_timeout_short(self):
1881
self.assert_command_from_args(["-b", "client"],
1882
command.BumpTimeout)
1884
def test_start_checker(self):
1885
self.assert_command_from_args(["--start-checker", "client"],
1886
command.StartChecker)
1888
def test_stop_checker(self):
1889
self.assert_command_from_args(["--stop-checker", "client"],
1890
command.StopChecker)
1892
def test_approve_by_default(self):
1893
self.assert_command_from_args(["--approve-by-default",
1895
command.ApproveByDefault)
1897
def test_deny_by_default(self):
1898
self.assert_command_from_args(["--deny-by-default", "client"],
1899
command.DenyByDefault)
1901
def test_checker(self):
1902
self.assert_command_from_args(["--checker", ":", "client"],
1906
def test_checker_empty(self):
1907
self.assert_command_from_args(["--checker", "", "client"],
1911
def test_checker_short(self):
1912
self.assert_command_from_args(["-c", ":", "client"],
1916
def test_host(self):
1917
self.assert_command_from_args(
1918
["--host", "client.example.org", "client"],
1919
command.SetHost, value_to_set="client.example.org")
1921
def test_host_short(self):
1922
self.assert_command_from_args(
1923
["-H", "client.example.org", "client"], command.SetHost,
1924
value_to_set="client.example.org")
1926
def test_secret_devnull(self):
1927
self.assert_command_from_args(["--secret", os.path.devnull,
1928
"client"], command.SetSecret,
1931
def test_secret_tempfile(self):
1932
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1933
value = b"secret\0xyzzy\nbar"
1936
self.assert_command_from_args(["--secret", f.name,
1941
def test_secret_devnull_short(self):
1942
self.assert_command_from_args(["-s", os.path.devnull,
1943
"client"], command.SetSecret,
1946
def test_secret_tempfile_short(self):
1947
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1948
value = b"secret\0xyzzy\nbar"
1951
self.assert_command_from_args(["-s", f.name, "client"],
1955
def test_timeout(self):
1956
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1958
value_to_set=300000)
1960
def test_timeout_short(self):
1961
self.assert_command_from_args(["-t", "PT5M", "client"],
1963
value_to_set=300000)
1965
def test_extended_timeout(self):
1966
self.assert_command_from_args(["--extended-timeout", "PT15M",
1968
command.SetExtendedTimeout,
1969
value_to_set=900000)
1971
def test_interval(self):
1972
self.assert_command_from_args(["--interval", "PT2M",
1973
"client"], command.SetInterval,
1974
value_to_set=120000)
1976
def test_interval_short(self):
1977
self.assert_command_from_args(["-i", "PT2M", "client"],
1978
command.SetInterval,
1979
value_to_set=120000)
1981
def test_approval_delay(self):
1982
self.assert_command_from_args(["--approval-delay", "PT30S",
1984
command.SetApprovalDelay,
1987
def test_approval_duration(self):
1988
self.assert_command_from_args(["--approval-duration", "PT1S",
1990
command.SetApprovalDuration,
1993
def test_print_table(self):
1994
self.assert_command_from_args([], command.PrintTable,
1997
def test_print_table_verbose(self):
1998
self.assert_command_from_args(["--verbose"],
2002
def test_print_table_verbose_short(self):
2003
self.assert_command_from_args(["-v"], command.PrintTable,
2007
def test_manual_page_example_1(self):
2008
self.assert_command_from_args("",
2013
def test_manual_page_example_2(self):
2014
self.assert_command_from_args(
2015
"--verbose foo1.example.org foo2.example.org".split(),
2016
command.PrintTable, clients=["foo1.example.org",
2017
"foo2.example.org"],
2020
def test_manual_page_example_3(self):
2021
self.assert_command_from_args("--enable --all".split(),
2025
def test_manual_page_example_4(self):
2026
self.assert_commands_from_args(
2027
("--timeout=PT5M --interval=PT1M foo1.example.org"
2028
" foo2.example.org").split(),
2029
[command.SetTimeout, command.SetInterval],
2030
clients=["foo1.example.org", "foo2.example.org"])
2032
def test_manual_page_example_5(self):
2033
self.assert_command_from_args("--approve --all".split(),
2038
class TestCommand(unittest.TestCase):
775
log.removeFilter(warning_filter)
776
self.assertTrue(getattr(warning_filter, "found", False))
777
self.assertEqual(value, datetime.timedelta(0, 7200))
780
class TestCmd(unittest.TestCase):
2039
781
"""Abstract class for tests of command classes"""
2041
class FakeMandosBus(dbus.MandosBus):
2042
def __init__(self, testcase):
2043
self.client_properties = {
784
class MockClient(object):
785
def __init__(self, name, **attributes):
786
self.__dbus_object_path__ = "objpath_{}".format(name)
787
self.attributes = attributes
788
self.attributes["Name"] = name
790
def Set(self, interface, property, value, dbus_interface):
791
testcase.assertEqual(interface, client_interface)
792
testcase.assertEqual(dbus_interface,
793
dbus.PROPERTIES_IFACE)
794
self.attributes[property] = value
795
def Get(self, interface, property, dbus_interface):
796
testcase.assertEqual(interface, client_interface)
797
testcase.assertEqual(dbus_interface,
798
dbus.PROPERTIES_IFACE)
799
return self.attributes[property]
800
def Approve(self, approve, dbus_interface):
801
testcase.assertEqual(dbus_interface, client_interface)
802
self.calls.append(("Approve", (approve,
804
self.client = MockClient(
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",
828
self.other_client = MockClient(
830
KeyID=("0558568eedd67d622f5c83b35a115f79"
831
"6ab612cff5ad227247e46c2b020f441c"),
834
Enabled=dbus.Boolean(True),
836
LastCheckedOK="2019-02-04T00:00:00",
837
Created="2019-01-03T00:00:00",
839
Fingerprint=("3E393AEAEFB84C7E89E2"
840
"F547B3A107558FCA3A27"),
841
CheckerRunning=dbus.Boolean(True),
842
LastEnabled="2019-01-04T00:00:00",
843
ApprovalPending=dbus.Boolean(False),
844
ApprovedByDefault=dbus.Boolean(False),
845
LastApprovalRequest="2019-01-03T00:00:00",
847
ApprovalDuration=1000,
849
ExtendedTimeout=900000,
850
Expires="2019-02-05T00:00:00",
851
LastCheckerStatus=-2)
852
self.clients = collections.OrderedDict(
854
(self.client, self.client.attributes),
855
(self.other_client, self.other_client.attributes),
857
self.one_client = {self.client: self.client.attributes}
859
class TestPrintTableCmd(TestCmd):
860
def test_normal(self):
861
output = PrintTableCmd().output(self.clients)
862
expected_output = """
863
Name Enabled Timeout Last Successful Check
864
foo Yes 00:05:00 2019-02-03T00:00:00
865
barbar Yes 00:05:00 2019-02-04T00:00:00
867
self.assertEqual(output, expected_output)
868
def test_verbose(self):
869
output = PrintTableCmd(verbose=True).output(self.clients)
870
expected_output = """
871
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
872
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
873
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
875
self.assertEqual(output, expected_output)
876
def test_one_client(self):
877
output = PrintTableCmd().output(self.one_client)
878
expected_output = """
879
Name Enabled Timeout Last Successful Check
880
foo Yes 00:05:00 2019-02-03T00:00:00
882
self.assertEqual(output, expected_output)
884
class TestDumpJSONCmd(TestCmd):
886
self.expected_json = {
2045
889
"KeyID": ("92ed150794387c03ce684574b1139a65"
2046
890
"94a34f895daaaf09fd8ea90a27cddb12"),
2047
"Secret": b"secret",
2048
891
"Host": "foo.example.org",
2049
892
"Enabled": True,
2050
893
"Timeout": 300000,
2084
926
"ApprovedByDefault": False,
2085
927
"LastApprovalRequest": "2019-01-03T00:00:00",
2086
928
"ApprovalDelay": 30000,
2087
"ApprovalDuration": 93785000,
929
"ApprovalDuration": 1000,
2089
931
"ExtendedTimeout": 900000,
2090
932
"Expires": "2019-02-05T00:00:00",
2091
933
"LastCheckerStatus": -2,
2093
self.clients = collections.OrderedDict(
2095
("client_objectpath", self.client_properties),
2096
("other_client_objectpath",
2097
self.other_client_properties),
2099
self.one_client = {"client_objectpath":
2100
self.client_properties}
2101
self.testcase = testcase
2104
def call_method(self, methodname, busname, objectpath,
2106
self.testcase.assertEqual("se.recompile.Mandos", busname)
2107
self.calls.append((methodname, busname, objectpath,
2109
if interface == "org.freedesktop.DBus.Properties":
2110
if methodname == "Set":
2111
self.testcase.assertEqual(3, len(args))
2112
interface, key, value = args
2113
self.testcase.assertEqual(
2114
"se.recompile.Mandos.Client", interface)
2115
self.clients[objectpath][key] = value
2117
elif interface == "se.recompile.Mandos":
2118
self.testcase.assertEqual("RemoveClient", methodname)
2119
self.testcase.assertEqual(1, len(args))
2120
clientpath = args[0]
2121
del self.clients[clientpath]
2123
elif interface == "se.recompile.Mandos.Client":
2124
if methodname == "Approve":
2125
self.testcase.assertEqual(1, len(args))
2130
self.bus = self.FakeMandosBus(self)
2133
class TestBaseCommands(TestCommand):
2135
def test_IsEnabled_exits_successfully(self):
936
return super(TestDumpJSONCmd, self).setUp()
937
def test_normal(self):
938
json_data = json.loads(DumpJSONCmd().output(self.clients))
939
self.assertDictEqual(json_data, self.expected_json)
940
def test_one_client(self):
941
clients = self.one_client
942
json_data = json.loads(DumpJSONCmd().output(clients))
943
expected_json = {"foo": self.expected_json["foo"]}
944
self.assertDictEqual(json_data, expected_json)
946
class TestIsEnabledCmd(TestCmd):
947
def test_is_enabled(self):
948
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
949
for client, properties in self.clients.items()))
950
def test_is_enabled_run_exits_successfully(self):
2136
951
with self.assertRaises(SystemExit) as e:
2137
command.IsEnabled().run(self.bus.one_client)
952
IsEnabledCmd().run(None, self.one_client)
2138
953
if e.exception.code is not None:
2139
self.assertEqual(0, e.exception.code)
954
self.assertEqual(e.exception.code, 0)
2141
956
self.assertIsNone(e.exception.code)
2143
def test_IsEnabled_exits_with_failure(self):
2144
self.bus.client_properties["Enabled"] = False
957
def test_is_enabled_run_exits_with_failure(self):
958
self.client.attributes["Enabled"] = dbus.Boolean(False)
2145
959
with self.assertRaises(SystemExit) as e:
2146
command.IsEnabled().run(self.bus.one_client)
960
IsEnabledCmd().run(None, self.one_client)
2147
961
if isinstance(e.exception.code, int):
2148
self.assertNotEqual(0, e.exception.code)
962
self.assertNotEqual(e.exception.code, 0)
2150
964
self.assertIsNotNone(e.exception.code)
2152
def test_Approve(self):
2153
busname = "se.recompile.Mandos"
2154
client_interface = "se.recompile.Mandos.Client"
2155
command.Approve().run(self.bus.clients, self.bus)
2156
for clientpath in self.bus.clients:
2157
self.assertIn(("Approve", busname, clientpath,
2158
client_interface, (True,)), self.bus.calls)
2160
def test_Deny(self):
2161
busname = "se.recompile.Mandos"
2162
client_interface = "se.recompile.Mandos.Client"
2163
command.Deny().run(self.bus.clients, self.bus)
2164
for clientpath in self.bus.clients:
2165
self.assertIn(("Approve", busname, clientpath,
2166
client_interface, (False,)),
2169
def test_Remove(self):
2170
command.Remove().run(self.bus.clients, self.bus)
2171
for clientpath in self.bus.clients:
2172
self.assertIn(("RemoveClient", dbus_busname,
2173
dbus_server_path, dbus_server_interface,
2174
(clientpath,)), self.bus.calls)
2179
"KeyID": ("92ed150794387c03ce684574b1139a65"
2180
"94a34f895daaaf09fd8ea90a27cddb12"),
2181
"Host": "foo.example.org",
2184
"LastCheckedOK": "2019-02-03T00:00:00",
2185
"Created": "2019-01-02T00:00:00",
2187
"Fingerprint": ("778827225BA7DE539C5A"
2188
"7CFA59CFF7CDBD9A5920"),
2189
"CheckerRunning": False,
2190
"LastEnabled": "2019-01-03T00:00:00",
2191
"ApprovalPending": False,
2192
"ApprovedByDefault": True,
2193
"LastApprovalRequest": "",
2195
"ApprovalDuration": 1000,
2196
"Checker": "fping -q -- %(host)s",
2197
"ExtendedTimeout": 900000,
2198
"Expires": "2019-02-04T00:00:00",
2199
"LastCheckerStatus": 0,
2203
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
2204
"6ab612cff5ad227247e46c2b020f441c"),
2205
"Host": "192.0.2.3",
2208
"LastCheckedOK": "2019-02-04T00:00:00",
2209
"Created": "2019-01-03T00:00:00",
2211
"Fingerprint": ("3E393AEAEFB84C7E89E2"
2212
"F547B3A107558FCA3A27"),
2213
"CheckerRunning": True,
2214
"LastEnabled": "2019-01-04T00:00:00",
2215
"ApprovalPending": False,
2216
"ApprovedByDefault": False,
2217
"LastApprovalRequest": "2019-01-03T00:00:00",
2218
"ApprovalDelay": 30000,
2219
"ApprovalDuration": 93785000,
2221
"ExtendedTimeout": 900000,
2222
"Expires": "2019-02-05T00:00:00",
2223
"LastCheckerStatus": -2,
2227
def test_DumpJSON_normal(self):
2228
with self.capture_stdout_to_buffer() as buffer:
2229
command.DumpJSON().run(self.bus.clients)
2230
json_data = json.loads(buffer.getvalue())
2231
self.assertDictEqual(self.expected_json, json_data)
2234
@contextlib.contextmanager
2235
def capture_stdout_to_buffer():
2236
capture_buffer = io.StringIO()
2237
old_stdout = sys.stdout
2238
sys.stdout = capture_buffer
2240
yield capture_buffer
2242
sys.stdout = old_stdout
2244
def test_DumpJSON_one_client(self):
2245
with self.capture_stdout_to_buffer() as buffer:
2246
command.DumpJSON().run(self.bus.one_client)
2247
json_data = json.loads(buffer.getvalue())
2248
expected_json = {"foo": self.expected_json["foo"]}
2249
self.assertDictEqual(expected_json, json_data)
2251
def test_PrintTable_normal(self):
2252
with self.capture_stdout_to_buffer() as buffer:
2253
command.PrintTable().run(self.bus.clients)
2254
expected_output = "\n".join((
2255
"Name Enabled Timeout Last Successful Check",
2256
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2257
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
2259
self.assertEqual(expected_output, buffer.getvalue())
2261
def test_PrintTable_verbose(self):
2262
with self.capture_stdout_to_buffer() as buffer:
2263
command.PrintTable(verbose=True).run(self.bus.clients)
2278
"Last Successful Check ",
2279
"2019-02-03T00:00:00 ",
2280
"2019-02-04T00:00:00 ",
2283
"2019-01-02T00:00:00 ",
2284
"2019-01-03T00:00:00 ",
2296
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
2298
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
2302
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
2303
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
2305
"Check Is Running ",
2310
"2019-01-03T00:00:00 ",
2311
"2019-01-04T00:00:00 ",
2313
"Approval Is Pending ",
2317
"Approved By Default ",
2321
"Last Approval Request ",
2323
"2019-01-03T00:00:00 ",
2329
"Approval Duration ",
2334
"fping -q -- %(host)s ",
2337
"Extended Timeout ",
2342
"2019-02-04T00:00:00 ",
2343
"2019-02-05T00:00:00 ",
2345
"Last Checker Status",
2350
num_lines = max(len(rows) for rows in columns)
2351
expected_output = ("\n".join("".join(rows[line]
2352
for rows in columns)
2353
for line in range(num_lines))
2355
self.assertEqual(expected_output, buffer.getvalue())
2357
def test_PrintTable_one_client(self):
2358
with self.capture_stdout_to_buffer() as buffer:
2359
command.PrintTable().run(self.bus.one_client)
2360
expected_output = "\n".join((
2361
"Name Enabled Timeout Last Successful Check",
2362
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2364
self.assertEqual(expected_output, buffer.getvalue())
2367
class TestPropertySetterCmd(TestCommand):
2368
"""Abstract class for tests of command.PropertySetter classes"""
966
class TestRemoveCmd(TestCmd):
967
def test_remove(self):
968
class MockMandos(object):
971
def RemoveClient(self, dbus_path):
972
self.calls.append(("RemoveClient", (dbus_path,)))
973
mandos = MockMandos()
974
super(TestRemoveCmd, self).setUp()
975
RemoveCmd().run(mandos, self.clients)
976
self.assertEqual(len(mandos.calls), 2)
977
for client in self.clients:
978
self.assertIn(("RemoveClient",
979
(client.__dbus_object_path__,)),
982
class TestApproveCmd(TestCmd):
983
def test_approve(self):
984
ApproveCmd().run(None, self.clients)
985
for client in self.clients:
986
self.assertIn(("Approve", (True, client_interface)),
989
class TestDenyCmd(TestCmd):
991
DenyCmd().run(None, self.clients)
992
for client in self.clients:
993
self.assertIn(("Approve", (False, client_interface)),
996
class TestEnableCmd(TestCmd):
997
def test_enable(self):
998
for client in self.clients:
999
client.attributes["Enabled"] = False
1001
EnableCmd().run(None, self.clients)
1003
for client in self.clients:
1004
self.assertTrue(client.attributes["Enabled"])
1006
class TestDisableCmd(TestCmd):
1007
def test_disable(self):
1008
DisableCmd().run(None, self.clients)
1010
for client in self.clients:
1011
self.assertFalse(client.attributes["Enabled"])
1013
class Unique(object):
1014
"""Class for objects which exist only to be unique objects, since
1015
unittest.mock.sentinel only exists in Python 3.3"""
1017
class TestPropertyCmd(TestCmd):
1018
"""Abstract class for tests of PropertyCmd classes"""
2370
1019
def runTest(self):
2371
1020
if not hasattr(self, "command"):
2372
return # Abstract TestCase class
2374
if hasattr(self, "values_to_set"):
2375
cmd_args = [(value,) for value in self.values_to_set]
2376
values_to_get = getattr(self, "values_to_get",
2379
cmd_args = [() for x in range(len(self.values_to_get))]
2380
values_to_get = self.values_to_get
2381
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2382
for clientpath in self.bus.clients:
2383
self.bus.clients[clientpath][self.propname] = (
2385
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2386
for clientpath in self.bus.clients:
2387
value = (self.bus.clients[clientpath]
1022
values_to_get = getattr(self, "values_to_get",
1024
for value_to_set, value_to_get in zip(self.values_to_set,
1026
for client in self.clients:
1027
old_value = client.attributes[self.property]
1028
self.assertNotIsInstance(old_value, Unique)
1029
client.attributes[self.property] = Unique()
1030
self.run_command(value_to_set, self.clients)
1031
for client in self.clients:
1032
value = client.attributes[self.property]
2389
1033
self.assertNotIsInstance(value, Unique)
2390
self.assertEqual(value_to_get, value)
2393
class TestEnableCmd(TestPropertySetterCmd):
2394
command = command.Enable
2395
propname = "Enabled"
2396
values_to_get = [True]
2399
class TestDisableCmd(TestPropertySetterCmd):
2400
command = command.Disable
2401
propname = "Enabled"
2402
values_to_get = [False]
2405
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2406
command = command.BumpTimeout
2407
propname = "LastCheckedOK"
2408
values_to_get = [""]
2411
class TestStartCheckerCmd(TestPropertySetterCmd):
2412
command = command.StartChecker
2413
propname = "CheckerRunning"
2414
values_to_get = [True]
2417
class TestStopCheckerCmd(TestPropertySetterCmd):
2418
command = command.StopChecker
2419
propname = "CheckerRunning"
2420
values_to_get = [False]
2423
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2424
command = command.ApproveByDefault
2425
propname = "ApprovedByDefault"
2426
values_to_get = [True]
2429
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2430
command = command.DenyByDefault
2431
propname = "ApprovedByDefault"
2432
values_to_get = [False]
2435
class TestSetCheckerCmd(TestPropertySetterCmd):
2436
command = command.SetChecker
2437
propname = "Checker"
1034
self.assertEqual(value, value_to_get)
1035
def run_command(self, value, clients):
1036
self.command().run(None, clients)
1038
class TestBumpTimeoutCmd(TestPropertyCmd):
1039
command = BumpTimeoutCmd
1040
property = "LastCheckedOK"
1041
values_to_set = [""]
1043
class TestStartCheckerCmd(TestPropertyCmd):
1044
command = StartCheckerCmd
1045
property = "CheckerRunning"
1046
values_to_set = [dbus.Boolean(True)]
1048
class TestStopCheckerCmd(TestPropertyCmd):
1049
command = StopCheckerCmd
1050
property = "CheckerRunning"
1051
values_to_set = [dbus.Boolean(False)]
1053
class TestApproveByDefaultCmd(TestPropertyCmd):
1054
command = ApproveByDefaultCmd
1055
property = "ApprovedByDefault"
1056
values_to_set = [dbus.Boolean(True)]
1058
class TestDenyByDefaultCmd(TestPropertyCmd):
1059
command = DenyByDefaultCmd
1060
property = "ApprovedByDefault"
1061
values_to_set = [dbus.Boolean(False)]
1063
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1064
"""Abstract class for tests of PropertyCmd classes using the
1065
ValueArgumentMixIn"""
1067
if type(self) is TestValueArgumentPropertyCmd:
1069
return super(TestValueArgumentPropertyCmd, self).runTest()
1070
def run_command(self, value, clients):
1071
self.command(value).run(None, clients)
1073
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1074
command = SetCheckerCmd
1075
property = "Checker"
2438
1076
values_to_set = ["", ":", "fping -q -- %s"]
2441
class TestSetHostCmd(TestPropertySetterCmd):
2442
command = command.SetHost
2444
values_to_set = ["192.0.2.3", "client.example.org"]
2447
class TestSetSecretCmd(TestPropertySetterCmd):
2448
command = command.SetSecret
2450
values_to_set = [io.BytesIO(b""),
2451
io.BytesIO(b"secret\0xyzzy\nbar")]
2452
values_to_get = [f.getvalue() for f in values_to_set]
2455
class TestSetTimeoutCmd(TestPropertySetterCmd):
2456
command = command.SetTimeout
2457
propname = "Timeout"
2458
values_to_set = [datetime.timedelta(),
2459
datetime.timedelta(minutes=5),
2460
datetime.timedelta(seconds=1),
2461
datetime.timedelta(weeks=1),
2462
datetime.timedelta(weeks=52)]
2463
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2466
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
2467
command = command.SetExtendedTimeout
2468
propname = "ExtendedTimeout"
2469
values_to_set = [datetime.timedelta(),
2470
datetime.timedelta(minutes=5),
2471
datetime.timedelta(seconds=1),
2472
datetime.timedelta(weeks=1),
2473
datetime.timedelta(weeks=52)]
2474
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2477
class TestSetIntervalCmd(TestPropertySetterCmd):
2478
command = command.SetInterval
2479
propname = "Interval"
2480
values_to_set = [datetime.timedelta(),
2481
datetime.timedelta(minutes=5),
2482
datetime.timedelta(seconds=1),
2483
datetime.timedelta(weeks=1),
2484
datetime.timedelta(weeks=52)]
2485
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2488
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
2489
command = command.SetApprovalDelay
2490
propname = "ApprovalDelay"
2491
values_to_set = [datetime.timedelta(),
2492
datetime.timedelta(minutes=5),
2493
datetime.timedelta(seconds=1),
2494
datetime.timedelta(weeks=1),
2495
datetime.timedelta(weeks=52)]
2496
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2499
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
2500
command = command.SetApprovalDuration
2501
propname = "ApprovalDuration"
2502
values_to_set = [datetime.timedelta(),
2503
datetime.timedelta(minutes=5),
2504
datetime.timedelta(seconds=1),
2505
datetime.timedelta(weeks=1),
2506
datetime.timedelta(weeks=52)]
2507
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1078
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1079
command = SetHostCmd
1081
values_to_set = ["192.0.2.3", "foo.example.org"]
1083
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1084
command = SetSecretCmd
1086
values_to_set = [b"", b"secret"]
1088
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1089
command = SetTimeoutCmd
1090
property = "Timeout"
1091
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1092
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1094
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1095
command = SetExtendedTimeoutCmd
1096
property = "ExtendedTimeout"
1097
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1098
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1100
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1101
command = SetIntervalCmd
1102
property = "Interval"
1103
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1106
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1107
command = SetApprovalDelayCmd
1108
property = "ApprovalDelay"
1109
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1112
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1113
command = SetApprovalDurationCmd
1114
property = "ApprovalDuration"
1115
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1118
class TestOptions(unittest.TestCase):
1120
self.parser = argparse.ArgumentParser()
1121
add_command_line_options(self.parser)
1122
def commands_from_args(self, args):
1123
self.options = self.parser.parse_args(args)
1124
return commands_from_options(self.options)
1125
def test_default_is_show_table(self):
1126
commands = self.commands_from_args([])
1127
self.assertEqual(len(commands), 1)
1128
command = commands[0]
1129
self.assertIsInstance(command, PrintTableCmd)
1130
self.assertEqual(command.verbose, False)
1131
def test_show_table_verbose(self):
1132
commands = self.commands_from_args(["--verbose"])
1133
self.assertEqual(len(commands), 1)
1134
command = commands[0]
1135
self.assertIsInstance(command, PrintTableCmd)
1136
self.assertEqual(command.verbose, True)
1137
def test_enable(self):
1138
commands = self.commands_from_args(["--enable", "foo"])
1139
self.assertEqual(len(commands), 1)
1140
command = commands[0]
1141
self.assertIsInstance(command, EnableCmd)
1142
self.assertEqual(self.options.client, ["foo"])