274
## Classes for commands.
276
# Abstract classes first
277
class Command(object):
278
"""Abstract class for commands"""
279
def run(self, clients, bus=None, mandos=None):
280
"""Normal commands should implement run_on_one_client(), but
281
commands which want to operate on all clients at the same time
282
can override this run() method instead."""
284
for clientpath, properties in clients.items():
285
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
286
busname, str(clientpath))
287
client = bus.get_object(busname, clientpath)
288
self.run_on_one_client(client, properties)
290
class PrintCmd(Command):
291
"""Abstract class for commands printing client details"""
292
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
293
"Created", "Interval", "Host", "KeyID",
294
"Fingerprint", "CheckerRunning", "LastEnabled",
295
"ApprovalPending", "ApprovedByDefault",
296
"LastApprovalRequest", "ApprovalDelay",
297
"ApprovalDuration", "Checker", "ExtendedTimeout",
298
"Expires", "LastCheckerStatus")
299
def run(self, clients, bus=None, mandos=None):
300
print(self.output(clients.values()))
301
def output(self, clients):
302
raise NotImplementedError()
304
class PropertyCmd(Command):
305
"""Abstract class for Actions for setting one client property"""
306
def run_on_one_client(self, client, properties):
307
"""Set the Client's D-Bus property"""
308
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
309
client.__dbus_object_path__,
310
dbus.PROPERTIES_IFACE, client_interface,
311
self.propname, self.value_to_set
312
if not isinstance(self.value_to_set, dbus.Boolean)
313
else bool(self.value_to_set))
314
client.Set(client_interface, self.propname, self.value_to_set,
315
dbus_interface=dbus.PROPERTIES_IFACE)
318
raise NotImplementedError()
320
class ValueArgumentMixIn(object):
321
"""Mixin class for commands taking a value as argument"""
322
def __init__(self, value):
323
self.value_to_set = value
325
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
326
"""Mixin class for commands taking a value argument as
329
def value_to_set(self):
332
def value_to_set(self, value):
333
"""When setting, convert value to a datetime.timedelta"""
334
self._vts = int(round(value.total_seconds() * 1000))
336
# Actual (non-abstract) command classes
338
class PrintTableCmd(PrintCmd):
339
def __init__(self, verbose=False):
340
self.verbose = verbose
342
def output(self, clients):
343
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
344
keywords = default_keywords
346
keywords = self.all_keywords
347
return str(self.TableOfClients(clients, keywords))
349
class TableOfClients(object):
352
"Enabled": "Enabled",
353
"Timeout": "Timeout",
354
"LastCheckedOK": "Last Successful Check",
355
"LastApprovalRequest": "Last Approval Request",
356
"Created": "Created",
357
"Interval": "Interval",
359
"Fingerprint": "Fingerprint",
361
"CheckerRunning": "Check Is Running",
362
"LastEnabled": "Last Enabled",
363
"ApprovalPending": "Approval Is Pending",
364
"ApprovedByDefault": "Approved By Default",
365
"ApprovalDelay": "Approval Delay",
366
"ApprovalDuration": "Approval Duration",
367
"Checker": "Checker",
368
"ExtendedTimeout": "Extended Timeout",
369
"Expires": "Expires",
370
"LastCheckerStatus": "Last Checker Status",
373
def __init__(self, clients, keywords, tableheaders=None):
374
self.clients = clients
375
self.keywords = keywords
376
if tableheaders is not None:
377
self.tableheaders = tableheaders
380
return "\n".join(self.rows())
382
if sys.version_info.major == 2:
383
__unicode__ = __str__
385
return str(self).encode(locale.getpreferredencoding())
388
format_string = self.row_formatting_string()
389
rows = [self.header_line(format_string)]
390
rows.extend(self.client_line(client, format_string)
391
for client in self.clients)
394
def row_formatting_string(self):
395
"Format string used to format table rows"
396
return " ".join("{{{key}:{width}}}".format(
397
width=max(len(self.tableheaders[key]),
398
*(len(self.string_from_client(client, key))
399
for client in self.clients)),
401
for key in self.keywords)
403
def string_from_client(self, client, key):
404
return self.valuetostring(client[key], key)
407
def valuetostring(value, keyword):
408
if isinstance(value, dbus.Boolean):
409
return "Yes" if value else "No"
410
if keyword in ("Timeout", "Interval", "ApprovalDelay",
411
"ApprovalDuration", "ExtendedTimeout"):
412
return milliseconds_to_string(value)
415
def header_line(self, format_string):
416
return format_string.format(**self.tableheaders)
418
def client_line(self, client, format_string):
419
return format_string.format(
420
**{key: self.string_from_client(client, key)
421
for key in self.keywords})
425
class DumpJSONCmd(PrintCmd):
426
def output(self, clients):
427
data = {client["Name"]:
428
{key: self.dbus_boolean_to_bool(client[key])
429
for key in self.all_keywords}
430
for client in clients.values()}
431
return json.dumps(data, indent=4, separators=(',', ': '))
433
def dbus_boolean_to_bool(value):
434
if isinstance(value, dbus.Boolean):
438
class IsEnabledCmd(Command):
439
def run(self, clients, bus=None, mandos=None):
440
client, properties = next(iter(clients.items()))
441
if self.is_enabled(client, properties):
444
def is_enabled(self, client, properties):
445
return properties["Enabled"]
447
class RemoveCmd(Command):
448
def run_on_one_client(self, client, properties):
449
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", busname,
450
server_path, server_interface,
451
str(client.__dbus_object_path__))
452
self.mandos.RemoveClient(client.__dbus_object_path__)
454
class ApproveCmd(Command):
455
def run_on_one_client(self, client, properties):
456
log.debug("D-Bus: %s:%s:%s.Approve(True)", busname,
457
client.__dbus_object_path__, client_interface)
458
client.Approve(dbus.Boolean(True),
459
dbus_interface=client_interface)
461
class DenyCmd(Command):
462
def run_on_one_client(self, client, properties):
463
log.debug("D-Bus: %s:%s:%s.Approve(False)", busname,
464
client.__dbus_object_path__, client_interface)
465
client.Approve(dbus.Boolean(False),
466
dbus_interface=client_interface)
468
class EnableCmd(PropertyCmd):
470
value_to_set = dbus.Boolean(True)
472
class DisableCmd(PropertyCmd):
474
value_to_set = dbus.Boolean(False)
476
class BumpTimeoutCmd(PropertyCmd):
477
propname = "LastCheckedOK"
480
class StartCheckerCmd(PropertyCmd):
481
propname = "CheckerRunning"
482
value_to_set = dbus.Boolean(True)
484
class StopCheckerCmd(PropertyCmd):
485
propname = "CheckerRunning"
486
value_to_set = dbus.Boolean(False)
488
class ApproveByDefaultCmd(PropertyCmd):
489
propname = "ApprovedByDefault"
490
value_to_set = dbus.Boolean(True)
492
class DenyByDefaultCmd(PropertyCmd):
493
propname = "ApprovedByDefault"
494
value_to_set = dbus.Boolean(False)
496
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
499
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
502
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
505
def value_to_set(self):
508
def value_to_set(self, value):
509
"""When setting, read data from supplied file object"""
510
self._vts = value.read()
513
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
516
class SetExtendedTimeoutCmd(PropertyCmd,
517
MillisecondsValueArgumentMixIn):
518
propname = "ExtendedTimeout"
520
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
521
propname = "Interval"
523
class SetApprovalDelayCmd(PropertyCmd,
524
MillisecondsValueArgumentMixIn):
525
propname = "ApprovalDelay"
527
class SetApprovalDurationCmd(PropertyCmd,
528
MillisecondsValueArgumentMixIn):
529
propname = "ApprovalDuration"
531
def add_command_line_options(parser):
532
parser.add_argument("--version", action="version",
533
version="%(prog)s {}".format(version),
534
help="show version number and exit")
535
parser.add_argument("-a", "--all", action="store_true",
536
help="Select all clients")
537
parser.add_argument("-v", "--verbose", action="store_true",
538
help="Print all fields")
539
parser.add_argument("-j", "--dump-json", action="store_true",
540
help="Dump client data in JSON format")
541
enable_disable = parser.add_mutually_exclusive_group()
542
enable_disable.add_argument("-e", "--enable", action="store_true",
543
help="Enable client")
544
enable_disable.add_argument("-d", "--disable",
546
help="disable client")
547
parser.add_argument("-b", "--bump-timeout", action="store_true",
548
help="Bump timeout for client")
549
start_stop_checker = parser.add_mutually_exclusive_group()
550
start_stop_checker.add_argument("--start-checker",
552
help="Start checker for client")
553
start_stop_checker.add_argument("--stop-checker",
555
help="Stop checker for client")
556
parser.add_argument("-V", "--is-enabled", action="store_true",
557
help="Check if client is enabled")
558
parser.add_argument("-r", "--remove", action="store_true",
559
help="Remove client")
560
parser.add_argument("-c", "--checker",
561
help="Set checker command for client")
562
parser.add_argument("-t", "--timeout", type=string_to_delta,
563
help="Set timeout for client")
564
parser.add_argument("--extended-timeout", type=string_to_delta,
565
help="Set extended timeout for client")
566
parser.add_argument("-i", "--interval", type=string_to_delta,
567
help="Set checker interval for client")
568
approve_deny_default = parser.add_mutually_exclusive_group()
569
approve_deny_default.add_argument(
570
"--approve-by-default", action="store_true",
571
default=None, dest="approved_by_default",
572
help="Set client to be approved by default")
573
approve_deny_default.add_argument(
574
"--deny-by-default", action="store_false",
575
dest="approved_by_default",
576
help="Set client to be denied by default")
577
parser.add_argument("--approval-delay", type=string_to_delta,
578
help="Set delay before client approve/deny")
579
parser.add_argument("--approval-duration", type=string_to_delta,
580
help="Set duration of one client approval")
581
parser.add_argument("-H", "--host", help="Set host for client")
582
parser.add_argument("-s", "--secret",
583
type=argparse.FileType(mode="rb"),
584
help="Set password blob (file) for client")
585
approve_deny = parser.add_mutually_exclusive_group()
586
approve_deny.add_argument(
587
"-A", "--approve", action="store_true",
588
help="Approve any current client request")
589
approve_deny.add_argument("-D", "--deny", action="store_true",
590
help="Deny any current client request")
591
parser.add_argument("--debug", action="store_true",
592
help="Debug mode (show D-Bus commands)")
593
parser.add_argument("--check", action="store_true",
594
help="Run self-test")
595
parser.add_argument("client", nargs="*", help="Client name")
598
def commands_from_options(options):
602
if options.dump_json:
603
commands.append(DumpJSONCmd())
606
commands.append(EnableCmd())
609
commands.append(DisableCmd())
611
if options.bump_timeout:
612
commands.append(BumpTimeoutCmd())
614
if options.start_checker:
615
commands.append(StartCheckerCmd())
617
if options.stop_checker:
618
commands.append(StopCheckerCmd())
620
if options.is_enabled:
621
commands.append(IsEnabledCmd())
623
if options.checker is not None:
624
commands.append(SetCheckerCmd(options.checker))
626
if options.timeout is not None:
627
commands.append(SetTimeoutCmd(options.timeout))
629
if options.extended_timeout:
631
SetExtendedTimeoutCmd(options.extended_timeout))
633
if options.interval is not None:
634
commands.append(SetIntervalCmd(options.interval))
636
if options.approved_by_default is not None:
637
if options.approved_by_default:
638
commands.append(ApproveByDefaultCmd())
640
commands.append(DenyByDefaultCmd())
642
if options.approval_delay is not None:
643
commands.append(SetApprovalDelayCmd(options.approval_delay))
645
if options.approval_duration is not None:
647
SetApprovalDurationCmd(options.approval_duration))
649
if options.host is not None:
650
commands.append(SetHostCmd(options.host))
652
if options.secret is not None:
653
commands.append(SetSecretCmd(options.secret))
656
commands.append(ApproveCmd())
659
commands.append(DenyCmd())
662
commands.append(RemoveCmd())
664
# If no command option has been given, show table of clients,
665
# optionally verbosely
667
commands.append(PrintTableCmd(verbose=options.verbose))
423
672
def check_option_syntax(parser, options):
424
673
"""Apply additional restrictions on options, not expressible in
427
def has_commands(options, commands=None):
429
commands = (command.Enable,
432
command.StartChecker,
438
command.SetExtendedTimeout,
440
command.ApproveByDefault,
441
command.DenyByDefault,
442
command.SetApprovalDelay,
443
command.SetApprovalDuration,
448
return any(isinstance(cmd, commands)
449
for cmd in options.commands)
676
def has_actions(options):
677
return any((options.enable,
679
options.bump_timeout,
680
options.start_checker,
681
options.stop_checker,
684
options.checker is not None,
685
options.timeout is not None,
686
options.extended_timeout is not None,
687
options.interval is not None,
688
options.approved_by_default is not None,
689
options.approval_delay is not None,
690
options.approval_duration is not None,
691
options.host is not None,
692
options.secret is not None,
451
if has_commands(options) and not (options.client or options.all):
696
if has_actions(options) and not (options.client or options.all):
452
697
parser.error("Options require clients names or --all.")
453
if options.verbose and has_commands(options):
698
if options.verbose and has_actions(options):
454
699
parser.error("--verbose can only be used alone.")
455
if (has_commands(options, (command.DumpJSON,))
456
and (options.verbose or len(options.commands) > 1)):
700
if options.dump_json and (options.verbose
701
or has_actions(options)):
457
702
parser.error("--dump-json can only be used alone.")
458
if options.all and not has_commands(options):
703
if options.all and not has_actions(options):
459
704
parser.error("--all requires an action.")
460
if (has_commands(options, (command.IsEnabled,))
461
and len(options.client) > 1):
705
if options.is_enabled and len(options.client) > 1:
462
706
parser.error("--is-enabled requires exactly one client")
463
if (len(options.commands) > 1
464
and has_commands(options, (command.Remove,))
465
and not has_commands(options, (command.Deny,))):
466
parser.error("--remove can only be combined with --deny")
471
class SystemBus(object):
473
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
474
def get_managed_objects(self, busname, objectpath):
475
return self.call_method("GetManagedObjects", busname,
477
self.object_manager_iface)
479
properties_iface = "org.freedesktop.DBus.Properties"
480
def set_property(self, busname, objectpath, interface, key,
482
self.call_method("Set", busname, objectpath,
483
self.properties_iface, interface, key,
487
class MandosBus(SystemBus):
488
busname_domain = "se.recompile"
489
busname = busname_domain + ".Mandos"
491
server_interface = busname_domain + ".Mandos"
492
client_interface = busname_domain + ".Mandos.Client"
495
def get_clients_and_properties(self):
496
managed_objects = self.get_managed_objects(
497
self.busname, self.server_path)
498
return {objpath: properties[self.client_interface]
499
for objpath, properties in managed_objects.items()
500
if self.client_interface in properties}
502
def set_client_property(self, objectpath, key, value):
503
return self.set_property(self.busname, objectpath,
504
self.client_interface, key,
507
def call_client_method(self, objectpath, method, *args):
508
return self.call_method(method, self.busname, objectpath,
509
self.client_interface, *args)
511
def call_server_method(self, method, *args):
512
return self.call_method(method, self.busname,
514
self.server_interface, *args)
516
class Error(Exception):
519
class ConnectFailed(Error):
523
class dbus_python_adapter(object):
525
class SystemBus(dbus.MandosBus):
526
"""Use dbus-python"""
528
def __init__(self, module=dbus_python):
529
self.dbus_python = module
530
self.bus = self.dbus_python.SystemBus()
532
@contextlib.contextmanager
533
def convert_exception(self, exception_class=dbus.Error):
536
except self.dbus_python.exceptions.DBusException as e:
537
# This does what "raise from" would do
538
exc = exception_class(*e.args)
542
def call_method(self, methodname, busname, objectpath,
544
proxy_object = self.get_object(busname, objectpath)
545
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
546
interface, methodname,
547
", ".join(repr(a) for a in args))
548
method = getattr(proxy_object, methodname)
549
with self.convert_exception():
550
with dbus_python_adapter.SilenceLogger(
552
value = method(*args, dbus_interface=interface)
553
return self.type_filter(value)
555
def get_object(self, busname, objectpath):
556
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
558
with self.convert_exception(dbus.ConnectFailed):
559
return self.bus.get_object(busname, objectpath)
561
def type_filter(self, value):
562
"""Convert the most bothersome types to Python types"""
563
if isinstance(value, self.dbus_python.Boolean):
565
if isinstance(value, self.dbus_python.ObjectPath):
567
# Also recurse into dictionaries
568
if isinstance(value, self.dbus_python.Dictionary):
569
return {self.type_filter(key):
570
self.type_filter(subval)
571
for key, subval in value.items()}
574
def set_client_property(self, objectpath, key, value):
576
if not isinstance(value, bytes):
577
value = value.encode("utf-8")
578
value = self.dbus_python.ByteArray(value)
579
return self.set_property(self.busname, objectpath,
580
self.client_interface, key,
583
class SilenceLogger(object):
584
"Simple context manager to silence a particular logger"
585
def __init__(self, loggername):
586
self.logger = logging.getLogger(loggername)
589
self.logger.addFilter(self.nullfilter)
591
class NullFilter(logging.Filter):
592
def filter(self, record):
595
nullfilter = NullFilter()
597
def __exit__(self, exc_type, exc_val, exc_tb):
598
self.logger.removeFilter(self.nullfilter)
601
class CachingBus(SystemBus):
602
"""A caching layer for dbus_python_adapter.SystemBus"""
603
def __init__(self, *args, **kwargs):
604
self.object_cache = {}
605
super(dbus_python_adapter.CachingBus,
606
self).__init__(*args, **kwargs)
607
def get_object(self, busname, objectpath):
609
return self.object_cache[(busname, objectpath)]
612
dbus_python_adapter.CachingBus,
613
self).get_object(busname, objectpath)
614
self.object_cache[(busname, objectpath)] = new_object
618
class pydbus_adapter(object):
619
class SystemBus(dbus.MandosBus):
620
def __init__(self, module=pydbus):
622
self.bus = self.pydbus.SystemBus()
624
@contextlib.contextmanager
625
def convert_exception(self, exception_class=dbus.Error):
628
except gi.repository.GLib.Error as e:
629
# This does what "raise from" would do
630
exc = exception_class(*e.args)
634
def call_method(self, methodname, busname, objectpath,
636
proxy_object = self.get(busname, objectpath)
637
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
638
interface, methodname,
639
", ".join(repr(a) for a in args))
640
method = getattr(proxy_object[interface], methodname)
641
with self.convert_exception():
644
def get(self, busname, objectpath):
645
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
647
with self.convert_exception(dbus.ConnectFailed):
648
if sys.version_info.major <= 2:
649
with warnings.catch_warnings():
650
warnings.filterwarnings(
651
"ignore", "", DeprecationWarning,
652
r"^xml\.etree\.ElementTree$")
653
return self.bus.get(busname, objectpath)
655
return self.bus.get(busname, objectpath)
657
def set_property(self, busname, objectpath, interface, key,
659
proxy_object = self.get(busname, objectpath)
660
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
661
objectpath, self.properties_iface, interface,
663
setattr(proxy_object[interface], key, value)
665
class CachingBus(SystemBus):
666
"""A caching layer for pydbus_adapter.SystemBus"""
667
def __init__(self, *args, **kwargs):
668
self.object_cache = {}
669
super(pydbus_adapter.CachingBus,
670
self).__init__(*args, **kwargs)
671
def get(self, busname, objectpath):
673
return self.object_cache[(busname, objectpath)]
675
new_object = (super(pydbus_adapter.CachingBus, self)
676
.get(busname, objectpath))
677
self.object_cache[(busname, objectpath)] = new_object
681
def commands_from_options(options):
683
commands = list(options.commands)
685
def find_cmd(cmd, commands):
687
for i, c in enumerate(commands):
688
if isinstance(c, cmd):
692
# If command.Remove is present, move any instances of command.Deny
693
# to occur ahead of command.Remove.
694
index_of_remove = find_cmd(command.Remove, commands)
695
before_remove = commands[:index_of_remove]
696
after_remove = commands[index_of_remove:]
698
for cmd in after_remove:
699
if isinstance(cmd, command.Deny):
700
before_remove.append(cmd)
702
cleaned_after.append(cmd)
703
if cleaned_after != after_remove:
704
commands = before_remove + cleaned_after
706
# If no command option has been given, show table of clients,
707
# optionally verbosely
709
commands.append(command.PrintTable(verbose=options.verbose))
714
class command(object):
715
"""A namespace for command classes"""
718
"""Abstract base class for commands"""
719
def run(self, clients, bus=None):
720
"""Normal commands should implement run_on_one_client(),
721
but commands which want to operate on all clients at the same time can
722
override this run() method instead.
725
for client, properties in clients.items():
726
self.run_on_one_client(client, properties)
729
class IsEnabled(Base):
730
def run(self, clients, bus=None):
731
properties = next(iter(clients.values()))
732
if properties["Enabled"]:
738
def run_on_one_client(self, client, properties):
739
self.bus.call_client_method(client, "Approve", True)
743
def run_on_one_client(self, client, properties):
744
self.bus.call_client_method(client, "Approve", False)
748
def run(self, clients, bus):
749
for clientpath in frozenset(clients.keys()):
750
bus.call_server_method("RemoveClient", clientpath)
754
"""Abstract class for commands outputting client details"""
755
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
756
"Created", "Interval", "Host", "KeyID",
757
"Fingerprint", "CheckerRunning",
758
"LastEnabled", "ApprovalPending",
759
"ApprovedByDefault", "LastApprovalRequest",
760
"ApprovalDelay", "ApprovalDuration",
761
"Checker", "ExtendedTimeout", "Expires",
765
class DumpJSON(Output):
766
def run(self, clients, bus=None):
767
data = {properties["Name"]:
768
{key: properties[key]
769
for key in self.all_keywords}
770
for properties in clients.values()}
771
print(json.dumps(data, indent=4, separators=(',', ': ')))
774
class PrintTable(Output):
775
def __init__(self, verbose=False):
776
self.verbose = verbose
778
def run(self, clients, bus=None):
779
default_keywords = ("Name", "Enabled", "Timeout",
781
keywords = default_keywords
783
keywords = self.all_keywords
784
print(self.TableOfClients(clients.values(), keywords))
786
class TableOfClients(object):
789
"Enabled": "Enabled",
790
"Timeout": "Timeout",
791
"LastCheckedOK": "Last Successful Check",
792
"LastApprovalRequest": "Last Approval Request",
793
"Created": "Created",
794
"Interval": "Interval",
796
"Fingerprint": "Fingerprint",
798
"CheckerRunning": "Check Is Running",
799
"LastEnabled": "Last Enabled",
800
"ApprovalPending": "Approval Is Pending",
801
"ApprovedByDefault": "Approved By Default",
802
"ApprovalDelay": "Approval Delay",
803
"ApprovalDuration": "Approval Duration",
804
"Checker": "Checker",
805
"ExtendedTimeout": "Extended Timeout",
806
"Expires": "Expires",
807
"LastCheckerStatus": "Last Checker Status",
810
def __init__(self, clients, keywords):
811
self.clients = clients
812
self.keywords = keywords
815
return "\n".join(self.rows())
817
if sys.version_info.major == 2:
818
__unicode__ = __str__
820
return str(self).encode(
821
locale.getpreferredencoding())
824
format_string = self.row_formatting_string()
825
rows = [self.header_line(format_string)]
826
rows.extend(self.client_line(client, format_string)
827
for client in self.clients)
830
def row_formatting_string(self):
831
"Format string used to format table rows"
832
return " ".join("{{{key}:{width}}}".format(
833
width=max(len(self.tableheaders[key]),
834
*(len(self.string_from_client(client,
836
for client in self.clients)),
838
for key in self.keywords)
840
def string_from_client(self, client, key):
841
return self.valuetostring(client[key], key)
844
def valuetostring(cls, value, keyword):
845
if isinstance(value, bool):
846
return "Yes" if value else "No"
847
if keyword in ("Timeout", "Interval", "ApprovalDelay",
848
"ApprovalDuration", "ExtendedTimeout"):
849
return cls.milliseconds_to_string(value)
852
def header_line(self, format_string):
853
return format_string.format(**self.tableheaders)
855
def client_line(self, client, format_string):
856
return format_string.format(
857
**{key: self.string_from_client(client, key)
858
for key in self.keywords})
861
def milliseconds_to_string(ms):
862
td = datetime.timedelta(0, 0, 0, ms)
863
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
864
.format(days="{}T".format(td.days)
866
hours=td.seconds // 3600,
867
minutes=(td.seconds % 3600) // 60,
868
seconds=td.seconds % 60))
871
class PropertySetter(Base):
872
"Abstract class for Actions for setting one client property"
874
def run_on_one_client(self, client, properties=None):
875
"""Set the Client's D-Bus property"""
876
self.bus.set_client_property(client, self.propname,
881
raise NotImplementedError()
884
class Enable(PropertySetter):
889
class Disable(PropertySetter):
894
class BumpTimeout(PropertySetter):
895
propname = "LastCheckedOK"
899
class StartChecker(PropertySetter):
900
propname = "CheckerRunning"
904
class StopChecker(PropertySetter):
905
propname = "CheckerRunning"
909
class ApproveByDefault(PropertySetter):
910
propname = "ApprovedByDefault"
914
class DenyByDefault(PropertySetter):
915
propname = "ApprovedByDefault"
919
class PropertySetterValue(PropertySetter):
920
"""Abstract class for PropertySetter recieving a value as
921
constructor argument instead of a class attribute."""
922
def __init__(self, value):
923
self.value_to_set = value
926
def argparse(cls, argtype):
928
return cls(argtype(arg))
931
class SetChecker(PropertySetterValue):
935
class SetHost(PropertySetterValue):
939
class SetSecret(PropertySetterValue):
943
def value_to_set(self):
947
def value_to_set(self, value):
948
"""When setting, read data from supplied file object"""
949
self._vts = value.read()
953
class PropertySetterValueMilliseconds(PropertySetterValue):
954
"""Abstract class for PropertySetterValue taking a value
955
argument as a datetime.timedelta() but should store it as
959
def value_to_set(self):
963
def value_to_set(self, value):
964
"When setting, convert value from a datetime.timedelta"
965
self._vts = int(round(value.total_seconds() * 1000))
968
class SetTimeout(PropertySetterValueMilliseconds):
972
class SetExtendedTimeout(PropertySetterValueMilliseconds):
973
propname = "ExtendedTimeout"
976
class SetInterval(PropertySetterValueMilliseconds):
977
propname = "Interval"
980
class SetApprovalDelay(PropertySetterValueMilliseconds):
981
propname = "ApprovalDelay"
984
class SetApprovalDuration(PropertySetterValueMilliseconds):
985
propname = "ApprovalDuration"
708
options.remove = False
709
if has_actions(options) and not options.deny:
710
parser.error("--remove can only be combined with --deny")
711
options.remove = True
715
parser = argparse.ArgumentParser()
717
add_command_line_options(parser)
719
options = parser.parse_args()
721
check_option_syntax(parser, options)
723
clientnames = options.client
726
log.setLevel(logging.DEBUG)
729
bus = dbus.SystemBus()
730
log.debug("D-Bus: Connect to: (busname=%r, path=%r)", busname,
732
mandos_dbus_objc = bus.get_object(busname, server_path)
733
except dbus.exceptions.DBusException:
734
log.critical("Could not connect to Mandos server")
737
mandos_serv = dbus.Interface(mandos_dbus_objc,
738
dbus_interface=server_interface)
739
mandos_serv_object_manager = dbus.Interface(
740
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
742
# Filter out log message from dbus module
743
dbus_logger = logging.getLogger("dbus.proxies")
744
class NullFilter(logging.Filter):
745
def filter(self, record):
747
dbus_filter = NullFilter()
749
dbus_logger.addFilter(dbus_filter)
750
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,
751
server_path, dbus.OBJECT_MANAGER_IFACE)
752
mandos_clients = {path: ifs_and_props[client_interface]
753
for path, ifs_and_props in
754
mandos_serv_object_manager
755
.GetManagedObjects().items()
756
if client_interface in ifs_and_props}
757
except dbus.exceptions.DBusException as e:
758
log.critical("Failed to access Mandos server through D-Bus:"
762
# restore dbus logger
763
dbus_logger.removeFilter(dbus_filter)
765
# Compile dict of (clients: properties) to process
769
clients = {objpath: properties
770
for objpath, properties in mandos_clients.items()}
772
for name in clientnames:
773
for objpath, properties in mandos_clients.items():
774
if properties["Name"] == name:
775
clients[objpath] = properties
778
log.critical("Client not found on server: %r", name)
781
# Run all commands on clients
782
commands = commands_from_options(options)
783
for command in commands:
784
command.run(clients, bus, mandos_serv)
989
class TestCaseWithAssertLogs(unittest.TestCase):
990
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
992
if not hasattr(unittest.TestCase, "assertLogs"):
993
@contextlib.contextmanager
994
def assertLogs(self, logger, level=logging.INFO):
995
capturing_handler = self.CapturingLevelHandler(level)
996
old_level = logger.level
997
old_propagate = logger.propagate
998
logger.addHandler(capturing_handler)
999
logger.setLevel(level)
1000
logger.propagate = False
1002
yield capturing_handler.watcher
1004
logger.propagate = old_propagate
1005
logger.removeHandler(capturing_handler)
1006
logger.setLevel(old_level)
1007
self.assertGreater(len(capturing_handler.watcher.records),
1010
class CapturingLevelHandler(logging.Handler):
1011
def __init__(self, level, *args, **kwargs):
1012
logging.Handler.__init__(self, *args, **kwargs)
1013
self.watcher = self.LoggingWatcher([], [])
1014
def emit(self, record):
1015
self.watcher.records.append(record)
1016
self.watcher.output.append(self.format(record))
1018
LoggingWatcher = collections.namedtuple("LoggingWatcher",
1023
class Unique(object):
1024
"""Class for objects which exist only to be unique objects, since
1025
unittest.mock.sentinel only exists in Python 3.3"""
1028
class Test_string_to_delta(TestCaseWithAssertLogs):
1029
# Just test basic RFC 3339 functionality here, the doc string for
1030
# rfc3339_duration_to_delta() already has more comprehensive
1031
# tests, which are run by doctest.
1033
def test_rfc3339_zero_seconds(self):
1034
self.assertEqual(datetime.timedelta(),
1035
string_to_delta("PT0S"))
1037
def test_rfc3339_zero_days(self):
1038
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
1040
def test_rfc3339_one_second(self):
1041
self.assertEqual(datetime.timedelta(0, 1),
1042
string_to_delta("PT1S"))
1044
def test_rfc3339_two_hours(self):
1045
self.assertEqual(datetime.timedelta(0, 7200),
1046
string_to_delta("PT2H"))
787
class Test_milliseconds_to_string(unittest.TestCase):
789
self.assertEqual(milliseconds_to_string(93785000),
791
def test_no_days(self):
792
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
793
def test_all_zero(self):
794
self.assertEqual(milliseconds_to_string(0), "00:00:00")
795
def test_no_fractional_seconds(self):
796
self.assertEqual(milliseconds_to_string(400), "00:00:00")
797
self.assertEqual(milliseconds_to_string(900), "00:00:00")
798
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
800
class Test_string_to_delta(unittest.TestCase):
801
def test_handles_basic_rfc3339(self):
802
self.assertEqual(string_to_delta("PT0S"),
803
datetime.timedelta())
804
self.assertEqual(string_to_delta("P0D"),
805
datetime.timedelta())
806
self.assertEqual(string_to_delta("PT1S"),
807
datetime.timedelta(0, 1))
808
self.assertEqual(string_to_delta("PT2H"),
809
datetime.timedelta(0, 7200))
1048
810
def test_falls_back_to_pre_1_6_1_with_warning(self):
1049
with self.assertLogs(log, logging.WARNING):
1050
value = string_to_delta("2h")
1051
self.assertEqual(datetime.timedelta(0, 7200), value)
1054
class Test_check_option_syntax(unittest.TestCase):
1056
self.parser = argparse.ArgumentParser()
1057
add_command_line_options(self.parser)
1059
def test_actions_requires_client_or_all(self):
1060
for action, value in self.actions.items():
1061
args = self.actionargs(action, value)
1062
with self.assertParseError():
1063
self.parse_args(args)
1065
# This mostly corresponds to the definition from has_commands() in
1066
# check_option_syntax()
1070
"--bump-timeout": None,
1071
"--start-checker": None,
1072
"--stop-checker": None,
1073
"--is-enabled": None,
1076
"--timeout": "PT0S",
1077
"--extended-timeout": "PT0S",
1078
"--interval": "PT0S",
1079
"--approve-by-default": None,
1080
"--deny-by-default": None,
1081
"--approval-delay": "PT0S",
1082
"--approval-duration": "PT0S",
1083
"--host": "hostname",
1084
"--secret": "/dev/null",
1090
def actionargs(action, value, *args):
1091
if value is not None:
1092
return [action, value] + list(args)
811
# assertLogs only exists in Python 3.4
812
if hasattr(self, "assertLogs"):
813
with self.assertLogs(log, logging.WARNING):
814
value = string_to_delta("2h")
1094
return [action] + list(args)
1096
@contextlib.contextmanager
1097
def assertParseError(self):
1098
with self.assertRaises(SystemExit) as e:
1099
with self.redirect_stderr_to_devnull():
1101
# Exit code from argparse is guaranteed to be "2". Reference:
1102
# https://docs.python.org/3/library
1103
# /argparse.html#exiting-methods
1104
self.assertEqual(2, e.exception.code)
1106
def parse_args(self, args):
1107
options = self.parser.parse_args(args)
1108
check_option_syntax(self.parser, options)
1111
@contextlib.contextmanager
1112
def redirect_stderr_to_devnull():
1113
old_stderr = sys.stderr
1114
with contextlib.closing(open(os.devnull, "w")) as null:
816
class WarningFilter(logging.Filter):
817
"""Don't show, but record the presence of, warnings"""
818
def filter(self, record):
819
is_warning = record.levelno >= logging.WARNING
820
self.found = is_warning or getattr(self, "found",
822
return not is_warning
823
warning_filter = WarningFilter()
824
log.addFilter(warning_filter)
826
value = string_to_delta("2h")
1119
sys.stderr = old_stderr
1121
def check_option_syntax(self, options):
1122
check_option_syntax(self.parser, options)
1124
def test_actions_all_conflicts_with_verbose(self):
1125
for action, value in self.actions.items():
1126
args = self.actionargs(action, value, "--all",
1128
with self.assertParseError():
1129
self.parse_args(args)
1131
def test_actions_with_client_conflicts_with_verbose(self):
1132
for action, value in self.actions.items():
1133
args = self.actionargs(action, value, "--verbose",
1135
with self.assertParseError():
1136
self.parse_args(args)
1138
def test_dump_json_conflicts_with_verbose(self):
1139
args = ["--dump-json", "--verbose"]
1140
with self.assertParseError():
1141
self.parse_args(args)
1143
def test_dump_json_conflicts_with_action(self):
1144
for action, value in self.actions.items():
1145
args = self.actionargs(action, value, "--dump-json")
1146
with self.assertParseError():
1147
self.parse_args(args)
1149
def test_all_can_not_be_alone(self):
1151
with self.assertParseError():
1152
self.parse_args(args)
1154
def test_all_is_ok_with_any_action(self):
1155
for action, value in self.actions.items():
1156
args = self.actionargs(action, value, "--all")
1157
self.parse_args(args)
1159
def test_any_action_is_ok_with_one_client(self):
1160
for action, value in self.actions.items():
1161
args = self.actionargs(action, value, "client")
1162
self.parse_args(args)
1164
def test_one_client_with_all_actions_except_is_enabled(self):
1165
for action, value in self.actions.items():
1166
if action == "--is-enabled":
1168
args = self.actionargs(action, value, "client")
1169
self.parse_args(args)
1171
def test_two_clients_with_all_actions_except_is_enabled(self):
1172
for action, value in self.actions.items():
1173
if action == "--is-enabled":
1175
args = self.actionargs(action, value, "client1",
1177
self.parse_args(args)
1179
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1180
for action, value in self.actions.items():
1181
if action == "--is-enabled":
1183
args = self.actionargs(action, value, "client1",
1185
self.parse_args(args)
1187
def test_is_enabled_fails_without_client(self):
1188
args = ["--is-enabled"]
1189
with self.assertParseError():
1190
self.parse_args(args)
1192
def test_is_enabled_fails_with_two_clients(self):
1193
args = ["--is-enabled", "client1", "client2"]
1194
with self.assertParseError():
1195
self.parse_args(args)
1197
def test_remove_can_only_be_combined_with_action_deny(self):
1198
for action, value in self.actions.items():
1199
if action in {"--remove", "--deny"}:
1201
args = self.actionargs(action, value, "--all",
1203
with self.assertParseError():
1204
self.parse_args(args)
1207
class Test_dbus_exceptions(unittest.TestCase):
1209
def test_dbus_ConnectFailed_is_Error(self):
1210
with self.assertRaises(dbus.Error):
1211
raise dbus.ConnectFailed()
1214
class Test_dbus_MandosBus(unittest.TestCase):
1216
class MockMandosBus(dbus.MandosBus):
1218
self._name = "se.recompile.Mandos"
1219
self._server_path = "/"
1220
self._server_interface = "se.recompile.Mandos"
1221
self._client_interface = "se.recompile.Mandos.Client"
1223
self.call_method_return = Unique()
1225
def call_method(self, methodname, busname, objectpath,
1227
self.calls.append((methodname, busname, objectpath,
1229
return self.call_method_return
1232
self.bus = self.MockMandosBus()
1234
def test_set_client_property(self):
1235
self.bus.set_client_property("objectpath", "key", "value")
1236
expected_call = ("Set", self.bus._name, "objectpath",
1237
"org.freedesktop.DBus.Properties",
1238
(self.bus._client_interface, "key", "value"))
1239
self.assertIn(expected_call, self.bus.calls)
1241
def test_call_client_method(self):
1242
ret = self.bus.call_client_method("objectpath", "methodname")
1243
self.assertIs(self.bus.call_method_return, ret)
1244
expected_call = ("methodname", self.bus._name, "objectpath",
1245
self.bus._client_interface, ())
1246
self.assertIn(expected_call, self.bus.calls)
1248
def test_call_client_method_with_args(self):
1249
args = (Unique(), Unique())
1250
ret = self.bus.call_client_method("objectpath", "methodname",
1252
self.assertIs(self.bus.call_method_return, ret)
1253
expected_call = ("methodname", self.bus._name, "objectpath",
1254
self.bus._client_interface,
1256
self.assertIn(expected_call, self.bus.calls)
1258
def test_get_clients_and_properties(self):
1261
self.bus._client_interface: {
1265
"irrelevant_interface": {
1266
"key": "othervalue",
1270
"other_objectpath": {
1271
"other_irrelevant_interface": {
1277
expected_clients_and_properties = {
1283
self.bus.call_method_return = managed_objects
1284
ret = self.bus.get_clients_and_properties()
1285
self.assertDictEqual(expected_clients_and_properties, ret)
1286
expected_call = ("GetManagedObjects", self.bus._name,
1287
self.bus._server_path,
1288
"org.freedesktop.DBus.ObjectManager", ())
1289
self.assertIn(expected_call, self.bus.calls)
1291
def test_call_server_method(self):
1292
ret = self.bus.call_server_method("methodname")
1293
self.assertIs(self.bus.call_method_return, ret)
1294
expected_call = ("methodname", self.bus._name,
1295
self.bus._server_path,
1296
self.bus._server_interface, ())
1297
self.assertIn(expected_call, self.bus.calls)
1299
def test_call_server_method_with_args(self):
1300
args = (Unique(), Unique())
1301
ret = self.bus.call_server_method("methodname", *args)
1302
self.assertIs(self.bus.call_method_return, ret)
1303
expected_call = ("methodname", self.bus._name,
1304
self.bus._server_path,
1305
self.bus._server_interface,
1307
self.assertIn(expected_call, self.bus.calls)
1310
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1312
def MockDBusPython_func(self, func):
1313
class mock_dbus_python(object):
1314
"""mock dbus-python module"""
1315
class exceptions(object):
1316
"""Pseudo-namespace"""
1317
class DBusException(Exception):
1319
class SystemBus(object):
1321
def get_object(busname, objectpath):
1322
DBusObject = collections.namedtuple(
1323
"DBusObject", ("methodname", "Set"))
1324
def method(*args, **kwargs):
1325
self.assertEqual({"dbus_interface":
1329
def set_property(interface, key, value,
1330
dbus_interface=None):
1332
"org.freedesktop.DBus.Properties",
1334
self.assertEqual("Secret", key)
1335
return func(interface, key, value,
1336
dbus_interface=dbus_interface)
1337
return DBusObject(methodname=method,
1339
class Boolean(object):
1340
def __init__(self, value):
1341
self.value = bool(value)
1344
if sys.version_info.major == 2:
1345
__nonzero__ = __bool__
1346
class ObjectPath(str):
1348
class Dictionary(dict):
1350
class ByteArray(bytes):
1352
return mock_dbus_python
1354
def call_method(self, bus, methodname, busname, objectpath,
1356
with self.assertLogs(log, logging.DEBUG):
1357
return bus.call_method(methodname, busname, objectpath,
1360
def test_call_method_returns(self):
1361
expected_method_return = Unique()
1362
method_args = (Unique(), Unique())
1364
self.assertEqual(len(method_args), len(args))
1365
for marg, arg in zip(method_args, args):
1366
self.assertIs(marg, arg)
1367
return expected_method_return
1368
mock_dbus_python = self.MockDBusPython_func(func)
1369
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1370
ret = self.call_method(bus, "methodname", "busname",
1371
"objectpath", "interface",
1373
self.assertIs(ret, expected_method_return)
1375
def test_call_method_filters_bool_true(self):
1377
return method_return
1378
mock_dbus_python = self.MockDBusPython_func(func)
1379
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1380
method_return = mock_dbus_python.Boolean(True)
1381
ret = self.call_method(bus, "methodname", "busname",
1382
"objectpath", "interface")
1383
self.assertTrue(ret)
1384
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1386
def test_call_method_filters_bool_false(self):
1388
return method_return
1389
mock_dbus_python = self.MockDBusPython_func(func)
1390
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1391
method_return = mock_dbus_python.Boolean(False)
1392
ret = self.call_method(bus, "methodname", "busname",
1393
"objectpath", "interface")
1394
self.assertFalse(ret)
1395
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1397
def test_call_method_filters_objectpath(self):
1399
return method_return
1400
mock_dbus_python = self.MockDBusPython_func(func)
1401
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1402
method_return = mock_dbus_python.ObjectPath("objectpath")
1403
ret = self.call_method(bus, "methodname", "busname",
1404
"objectpath", "interface")
1405
self.assertEqual("objectpath", ret)
1406
self.assertIsNot("objectpath", ret)
1407
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1409
def test_call_method_filters_booleans_in_dict(self):
1411
return method_return
1412
mock_dbus_python = self.MockDBusPython_func(func)
1413
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1414
method_return = mock_dbus_python.Dictionary(
1415
{mock_dbus_python.Boolean(True):
1416
mock_dbus_python.Boolean(False),
1417
mock_dbus_python.Boolean(False):
1418
mock_dbus_python.Boolean(True)})
1419
ret = self.call_method(bus, "methodname", "busname",
1420
"objectpath", "interface")
1421
expected_method_return = {True: False,
1423
self.assertEqual(expected_method_return, ret)
1424
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1426
def test_call_method_filters_objectpaths_in_dict(self):
1428
return method_return
1429
mock_dbus_python = self.MockDBusPython_func(func)
1430
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1431
method_return = mock_dbus_python.Dictionary(
1432
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1433
mock_dbus_python.ObjectPath("objectpath_value_1"),
1434
mock_dbus_python.ObjectPath("objectpath_key_2"):
1435
mock_dbus_python.ObjectPath("objectpath_value_2")})
1436
ret = self.call_method(bus, "methodname", "busname",
1437
"objectpath", "interface")
1438
expected_method_return = {str(key): str(value)
1440
method_return.items()}
1441
self.assertEqual(expected_method_return, ret)
1442
self.assertIsInstance(ret, dict)
1443
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1445
def test_call_method_filters_dict_in_dict(self):
1447
return method_return
1448
mock_dbus_python = self.MockDBusPython_func(func)
1449
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1450
method_return = mock_dbus_python.Dictionary(
1451
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1452
"key12": "value12"}),
1453
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1454
"key22": "value22"})})
1455
ret = self.call_method(bus, "methodname", "busname",
1456
"objectpath", "interface")
1457
expected_method_return = {
1458
"key1": {"key11": "value11",
1459
"key12": "value12"},
1460
"key2": {"key21": "value21",
1461
"key22": "value22"},
1463
self.assertEqual(expected_method_return, ret)
1464
self.assertIsInstance(ret, dict)
1465
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1466
for key, value in ret.items():
1467
self.assertIsInstance(value, dict)
1468
self.assertEqual(expected_method_return[key], value)
1469
self.assertNotIsInstance(value,
1470
mock_dbus_python.Dictionary)
1472
def test_call_method_filters_dict_three_deep(self):
1474
return method_return
1475
mock_dbus_python = self.MockDBusPython_func(func)
1476
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1477
method_return = mock_dbus_python.Dictionary(
1479
mock_dbus_python.Dictionary(
1481
mock_dbus_python.Dictionary(
1483
mock_dbus_python.Boolean(True),
1487
ret = self.call_method(bus, "methodname", "busname",
1488
"objectpath", "interface")
1489
expected_method_return = {"key1": {"key2": {"key3": True}}}
1490
self.assertEqual(expected_method_return, ret)
1491
self.assertIsInstance(ret, dict)
1492
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1493
self.assertIsInstance(ret["key1"], dict)
1494
self.assertNotIsInstance(ret["key1"],
1495
mock_dbus_python.Dictionary)
1496
self.assertIsInstance(ret["key1"]["key2"], dict)
1497
self.assertNotIsInstance(ret["key1"]["key2"],
1498
mock_dbus_python.Dictionary)
1499
self.assertTrue(ret["key1"]["key2"]["key3"])
1500
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1501
mock_dbus_python.Boolean)
1503
def test_call_method_handles_exception(self):
1504
dbus_logger = logging.getLogger("dbus.proxies")
1507
dbus_logger.error("Test")
1508
raise mock_dbus_python.exceptions.DBusException()
1510
mock_dbus_python = self.MockDBusPython_func(func)
1511
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1513
class CountingHandler(logging.Handler):
1515
def emit(self, record):
1518
counting_handler = CountingHandler()
1520
dbus_logger.addHandler(counting_handler)
1523
with self.assertRaises(dbus.Error) as e:
1524
self.call_method(bus, "methodname", "busname",
1525
"objectpath", "interface")
1527
dbus_logger.removeFilter(counting_handler)
1529
self.assertNotIsInstance(e, dbus.ConnectFailed)
1531
# Make sure the dbus logger was suppressed
1532
self.assertEqual(0, counting_handler.count)
1534
def test_Set_Secret_sends_bytearray(self):
1536
def func(*args, **kwargs):
1537
ret[0] = (args, kwargs)
1538
mock_dbus_python = self.MockDBusPython_func(func)
1539
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1540
bus.set_client_property("objectpath", "Secret", "value")
1541
expected_call = (("se.recompile.Mandos.Client", "Secret",
1542
mock_dbus_python.ByteArray(b"value")),
1544
"org.freedesktop.DBus.Properties"})
1545
self.assertEqual(expected_call, ret[0])
1546
if sys.version_info.major == 2:
1547
self.assertIsInstance(ret[0][0][-1],
1548
mock_dbus_python.ByteArray)
1550
def test_get_object_converts_to_correct_exception(self):
1551
bus = dbus_python_adapter.SystemBus(
1552
self.fake_dbus_python_raises_exception_on_connect)
1553
with self.assertRaises(dbus.ConnectFailed):
1554
self.call_method(bus, "methodname", "busname",
1555
"objectpath", "interface")
1557
class fake_dbus_python_raises_exception_on_connect(object):
1558
"""fake dbus-python module"""
1559
class exceptions(object):
1560
"""Pseudo-namespace"""
1561
class DBusException(Exception):
1566
def get_object(busname, objectpath):
1567
raise cls.exceptions.DBusException()
1568
Bus = collections.namedtuple("Bus", ["get_object"])
1569
return Bus(get_object=get_object)
1572
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1573
class mock_dbus_python(object):
1574
"""mock dbus-python modules"""
1575
class SystemBus(object):
1577
def get_object(busname, objectpath):
1581
self.bus = dbus_python_adapter.CachingBus(
1582
self.mock_dbus_python)
1584
def test_returns_distinct_objectpaths(self):
1585
obj1 = self.bus.get_object("busname", "objectpath1")
1586
self.assertIsInstance(obj1, Unique)
1587
obj2 = self.bus.get_object("busname", "objectpath2")
1588
self.assertIsInstance(obj2, Unique)
1589
self.assertIsNot(obj1, obj2)
1591
def test_returns_distinct_busnames(self):
1592
obj1 = self.bus.get_object("busname1", "objectpath")
1593
self.assertIsInstance(obj1, Unique)
1594
obj2 = self.bus.get_object("busname2", "objectpath")
1595
self.assertIsInstance(obj2, Unique)
1596
self.assertIsNot(obj1, obj2)
1598
def test_returns_distinct_both(self):
1599
obj1 = self.bus.get_object("busname1", "objectpath")
1600
self.assertIsInstance(obj1, Unique)
1601
obj2 = self.bus.get_object("busname2", "objectpath")
1602
self.assertIsInstance(obj2, Unique)
1603
self.assertIsNot(obj1, obj2)
1605
def test_returns_same(self):
1606
obj1 = self.bus.get_object("busname", "objectpath")
1607
self.assertIsInstance(obj1, Unique)
1608
obj2 = self.bus.get_object("busname", "objectpath")
1609
self.assertIsInstance(obj2, Unique)
1610
self.assertIs(obj1, obj2)
1612
def test_returns_same_old(self):
1613
obj1 = self.bus.get_object("busname1", "objectpath1")
1614
self.assertIsInstance(obj1, Unique)
1615
obj2 = self.bus.get_object("busname2", "objectpath2")
1616
self.assertIsInstance(obj2, Unique)
1617
obj1b = self.bus.get_object("busname1", "objectpath1")
1618
self.assertIsInstance(obj1b, Unique)
1619
self.assertIsNot(obj1, obj2)
1620
self.assertIsNot(obj2, obj1b)
1621
self.assertIs(obj1, obj1b)
1624
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1626
def Stub_pydbus_func(self, func):
1627
class stub_pydbus(object):
1628
"""stub pydbus module"""
1629
class SystemBus(object):
1631
def get(busname, objectpath):
1632
DBusObject = collections.namedtuple(
1633
"DBusObject", ("methodname",))
1634
return {"interface":
1635
DBusObject(methodname=func)}
1638
def call_method(self, bus, methodname, busname, objectpath,
1640
with self.assertLogs(log, logging.DEBUG):
1641
return bus.call_method(methodname, busname, objectpath,
1644
def test_call_method_returns(self):
1645
expected_method_return = Unique()
1646
method_args = (Unique(), Unique())
1648
self.assertEqual(len(method_args), len(args))
1649
for marg, arg in zip(method_args, args):
1650
self.assertIs(marg, arg)
1651
return expected_method_return
1652
stub_pydbus = self.Stub_pydbus_func(func)
1653
bus = pydbus_adapter.SystemBus(stub_pydbus)
1654
ret = self.call_method(bus, "methodname", "busname",
1655
"objectpath", "interface",
1657
self.assertIs(ret, expected_method_return)
1659
def test_call_method_handles_exception(self):
1660
dbus_logger = logging.getLogger("dbus.proxies")
1663
raise gi.repository.GLib.Error()
1665
stub_pydbus = self.Stub_pydbus_func(func)
1666
bus = pydbus_adapter.SystemBus(stub_pydbus)
1668
with self.assertRaises(dbus.Error) as e:
1669
self.call_method(bus, "methodname", "busname",
1670
"objectpath", "interface")
1672
self.assertNotIsInstance(e, dbus.ConnectFailed)
1674
def test_get_converts_to_correct_exception(self):
1675
bus = pydbus_adapter.SystemBus(
1676
self.fake_pydbus_raises_exception_on_connect)
1677
with self.assertRaises(dbus.ConnectFailed):
1678
self.call_method(bus, "methodname", "busname",
1679
"objectpath", "interface")
1681
class fake_pydbus_raises_exception_on_connect(object):
1682
"""fake dbus-python module"""
1685
def get(busname, objectpath):
1686
raise gi.repository.GLib.Error()
1687
Bus = collections.namedtuple("Bus", ["get"])
1690
def test_set_property_uses_setattr(self):
1691
class Object(object):
1694
class pydbus_spy(object):
1695
class SystemBus(object):
1697
def get(busname, objectpath):
1698
return {"interface": obj}
1699
bus = pydbus_adapter.SystemBus(pydbus_spy)
1701
bus.set_property("busname", "objectpath", "interface", "key",
1703
self.assertIs(value, obj.key)
1705
def test_get_suppresses_xml_deprecation_warning(self):
1706
if sys.version_info.major >= 3:
1708
class stub_pydbus_get(object):
1709
class SystemBus(object):
1711
def get(busname, objectpath):
1712
warnings.warn_explicit(
1713
"deprecated", DeprecationWarning,
1714
"xml.etree.ElementTree", 0)
1715
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1716
with warnings.catch_warnings(record=True) as w:
1717
warnings.simplefilter("always")
1718
bus.get("busname", "objectpath")
1719
self.assertEqual(0, len(w))
1722
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1723
class stub_pydbus(object):
1724
"""stub pydbus module"""
1725
class SystemBus(object):
1727
def get(busname, objectpath):
1731
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1733
def test_returns_distinct_objectpaths(self):
1734
obj1 = self.bus.get("busname", "objectpath1")
1735
self.assertIsInstance(obj1, Unique)
1736
obj2 = self.bus.get("busname", "objectpath2")
1737
self.assertIsInstance(obj2, Unique)
1738
self.assertIsNot(obj1, obj2)
1740
def test_returns_distinct_busnames(self):
1741
obj1 = self.bus.get("busname1", "objectpath")
1742
self.assertIsInstance(obj1, Unique)
1743
obj2 = self.bus.get("busname2", "objectpath")
1744
self.assertIsInstance(obj2, Unique)
1745
self.assertIsNot(obj1, obj2)
1747
def test_returns_distinct_both(self):
1748
obj1 = self.bus.get("busname1", "objectpath")
1749
self.assertIsInstance(obj1, Unique)
1750
obj2 = self.bus.get("busname2", "objectpath")
1751
self.assertIsInstance(obj2, Unique)
1752
self.assertIsNot(obj1, obj2)
1754
def test_returns_same(self):
1755
obj1 = self.bus.get("busname", "objectpath")
1756
self.assertIsInstance(obj1, Unique)
1757
obj2 = self.bus.get("busname", "objectpath")
1758
self.assertIsInstance(obj2, Unique)
1759
self.assertIs(obj1, obj2)
1761
def test_returns_same_old(self):
1762
obj1 = self.bus.get("busname1", "objectpath1")
1763
self.assertIsInstance(obj1, Unique)
1764
obj2 = self.bus.get("busname2", "objectpath2")
1765
self.assertIsInstance(obj2, Unique)
1766
obj1b = self.bus.get("busname1", "objectpath1")
1767
self.assertIsInstance(obj1b, Unique)
1768
self.assertIsNot(obj1, obj2)
1769
self.assertIsNot(obj2, obj1b)
1770
self.assertIs(obj1, obj1b)
1773
class Test_commands_from_options(unittest.TestCase):
1776
self.parser = argparse.ArgumentParser()
1777
add_command_line_options(self.parser)
1779
def test_is_enabled(self):
1780
self.assert_command_from_args(["--is-enabled", "client"],
1783
def assert_command_from_args(self, args, command_cls,
1785
"""Assert that parsing ARGS should result in an instance of
1786
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1787
options = self.parser.parse_args(args)
1788
check_option_syntax(self.parser, options)
1789
commands = commands_from_options(options)
1790
self.assertEqual(1, len(commands))
1791
command = commands[0]
1792
self.assertIsInstance(command, command_cls)
1793
for key, value in cmd_attrs.items():
1794
self.assertEqual(value, getattr(command, key))
1796
def test_is_enabled_short(self):
1797
self.assert_command_from_args(["-V", "client"],
1800
def test_approve(self):
1801
self.assert_command_from_args(["--approve", "client"],
1804
def test_approve_short(self):
1805
self.assert_command_from_args(["-A", "client"],
1808
def test_deny(self):
1809
self.assert_command_from_args(["--deny", "client"],
1812
def test_deny_short(self):
1813
self.assert_command_from_args(["-D", "client"], command.Deny)
1815
def test_remove(self):
1816
self.assert_command_from_args(["--remove", "client"],
1819
def test_deny_before_remove(self):
1820
options = self.parser.parse_args(["--deny", "--remove",
1822
check_option_syntax(self.parser, options)
1823
commands = commands_from_options(options)
1824
self.assertEqual(2, len(commands))
1825
self.assertIsInstance(commands[0], command.Deny)
1826
self.assertIsInstance(commands[1], command.Remove)
1828
def test_deny_before_remove_reversed(self):
1829
options = self.parser.parse_args(["--remove", "--deny",
1831
check_option_syntax(self.parser, options)
1832
commands = commands_from_options(options)
1833
self.assertEqual(2, len(commands))
1834
self.assertIsInstance(commands[0], command.Deny)
1835
self.assertIsInstance(commands[1], command.Remove)
1837
def test_remove_short(self):
1838
self.assert_command_from_args(["-r", "client"],
1841
def test_dump_json(self):
1842
self.assert_command_from_args(["--dump-json"],
1845
def test_enable(self):
1846
self.assert_command_from_args(["--enable", "client"],
1849
def test_enable_short(self):
1850
self.assert_command_from_args(["-e", "client"],
1853
def test_disable(self):
1854
self.assert_command_from_args(["--disable", "client"],
1857
def test_disable_short(self):
1858
self.assert_command_from_args(["-d", "client"],
1861
def test_bump_timeout(self):
1862
self.assert_command_from_args(["--bump-timeout", "client"],
1863
command.BumpTimeout)
1865
def test_bump_timeout_short(self):
1866
self.assert_command_from_args(["-b", "client"],
1867
command.BumpTimeout)
1869
def test_start_checker(self):
1870
self.assert_command_from_args(["--start-checker", "client"],
1871
command.StartChecker)
1873
def test_stop_checker(self):
1874
self.assert_command_from_args(["--stop-checker", "client"],
1875
command.StopChecker)
1877
def test_approve_by_default(self):
1878
self.assert_command_from_args(["--approve-by-default",
1880
command.ApproveByDefault)
1882
def test_deny_by_default(self):
1883
self.assert_command_from_args(["--deny-by-default", "client"],
1884
command.DenyByDefault)
1886
def test_checker(self):
1887
self.assert_command_from_args(["--checker", ":", "client"],
1891
def test_checker_empty(self):
1892
self.assert_command_from_args(["--checker", "", "client"],
1896
def test_checker_short(self):
1897
self.assert_command_from_args(["-c", ":", "client"],
1901
def test_host(self):
1902
self.assert_command_from_args(
1903
["--host", "client.example.org", "client"],
1904
command.SetHost, value_to_set="client.example.org")
1906
def test_host_short(self):
1907
self.assert_command_from_args(
1908
["-H", "client.example.org", "client"], command.SetHost,
1909
value_to_set="client.example.org")
1911
def test_secret_devnull(self):
1912
self.assert_command_from_args(["--secret", os.path.devnull,
1913
"client"], command.SetSecret,
1916
def test_secret_tempfile(self):
1917
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1918
value = b"secret\0xyzzy\nbar"
1921
self.assert_command_from_args(["--secret", f.name,
1926
def test_secret_devnull_short(self):
1927
self.assert_command_from_args(["-s", os.path.devnull,
1928
"client"], command.SetSecret,
1931
def test_secret_tempfile_short(self):
1932
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1933
value = b"secret\0xyzzy\nbar"
1936
self.assert_command_from_args(["-s", f.name, "client"],
1940
def test_timeout(self):
1941
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1943
value_to_set=300000)
1945
def test_timeout_short(self):
1946
self.assert_command_from_args(["-t", "PT5M", "client"],
1948
value_to_set=300000)
1950
def test_extended_timeout(self):
1951
self.assert_command_from_args(["--extended-timeout", "PT15M",
1953
command.SetExtendedTimeout,
1954
value_to_set=900000)
1956
def test_interval(self):
1957
self.assert_command_from_args(["--interval", "PT2M",
1958
"client"], command.SetInterval,
1959
value_to_set=120000)
1961
def test_interval_short(self):
1962
self.assert_command_from_args(["-i", "PT2M", "client"],
1963
command.SetInterval,
1964
value_to_set=120000)
1966
def test_approval_delay(self):
1967
self.assert_command_from_args(["--approval-delay", "PT30S",
1969
command.SetApprovalDelay,
1972
def test_approval_duration(self):
1973
self.assert_command_from_args(["--approval-duration", "PT1S",
1975
command.SetApprovalDuration,
1978
def test_print_table(self):
1979
self.assert_command_from_args([], command.PrintTable,
1982
def test_print_table_verbose(self):
1983
self.assert_command_from_args(["--verbose"],
1987
def test_print_table_verbose_short(self):
1988
self.assert_command_from_args(["-v"], command.PrintTable,
1992
class TestCommand(unittest.TestCase):
828
log.removeFilter(warning_filter)
829
self.assertTrue(getattr(warning_filter, "found", False))
830
self.assertEqual(value, datetime.timedelta(0, 7200))
833
class TestCmd(unittest.TestCase):
1993
834
"""Abstract class for tests of command classes"""
1995
class FakeMandosBus(dbus.MandosBus):
1996
def __init__(self, testcase):
1997
self.client_properties = {
837
class MockClient(object):
838
def __init__(self, name, **attributes):
839
self.__dbus_object_path__ = "/clients/{}".format(name)
840
self.attributes = attributes
841
self.attributes["Name"] = name
843
def Set(self, interface, propname, value, dbus_interface):
844
testcase.assertEqual(interface, client_interface)
845
testcase.assertEqual(dbus_interface,
846
dbus.PROPERTIES_IFACE)
847
self.attributes[propname] = value
848
def Get(self, interface, propname, dbus_interface):
849
testcase.assertEqual(interface, client_interface)
850
testcase.assertEqual(dbus_interface,
851
dbus.PROPERTIES_IFACE)
852
return self.attributes[propname]
853
def Approve(self, approve, dbus_interface):
854
testcase.assertEqual(dbus_interface, client_interface)
855
self.calls.append(("Approve", (approve,
857
self.client = MockClient(
859
KeyID=("92ed150794387c03ce684574b1139a65"
860
"94a34f895daaaf09fd8ea90a27cddb12"),
862
Host="foo.example.org",
863
Enabled=dbus.Boolean(True),
865
LastCheckedOK="2019-02-03T00:00:00",
866
Created="2019-01-02T00:00:00",
868
Fingerprint=("778827225BA7DE539C5A"
869
"7CFA59CFF7CDBD9A5920"),
870
CheckerRunning=dbus.Boolean(False),
871
LastEnabled="2019-01-03T00:00:00",
872
ApprovalPending=dbus.Boolean(False),
873
ApprovedByDefault=dbus.Boolean(True),
874
LastApprovalRequest="",
876
ApprovalDuration=1000,
877
Checker="fping -q -- %(host)s",
878
ExtendedTimeout=900000,
879
Expires="2019-02-04T00:00:00",
881
self.other_client = MockClient(
883
KeyID=("0558568eedd67d622f5c83b35a115f79"
884
"6ab612cff5ad227247e46c2b020f441c"),
887
Enabled=dbus.Boolean(True),
889
LastCheckedOK="2019-02-04T00:00:00",
890
Created="2019-01-03T00:00:00",
892
Fingerprint=("3E393AEAEFB84C7E89E2"
893
"F547B3A107558FCA3A27"),
894
CheckerRunning=dbus.Boolean(True),
895
LastEnabled="2019-01-04T00:00:00",
896
ApprovalPending=dbus.Boolean(False),
897
ApprovedByDefault=dbus.Boolean(False),
898
LastApprovalRequest="2019-01-03T00:00:00",
900
ApprovalDuration=1000,
902
ExtendedTimeout=900000,
903
Expires="2019-02-05T00:00:00",
904
LastCheckerStatus=-2)
905
self.clients = collections.OrderedDict(
907
("/clients/foo", self.client.attributes),
908
("/clients/barbar", self.other_client.attributes),
910
self.one_client = {"/clients/foo": self.client.attributes}
915
def get_object(client_bus_name, path):
916
self.assertEqual(client_bus_name, busname)
918
"/clients/foo": self.client,
919
"/clients/barbar": self.other_client,
923
class TestPrintTableCmd(TestCmd):
924
def test_normal(self):
925
output = PrintTableCmd().output(self.clients.values())
926
expected_output = """
927
Name Enabled Timeout Last Successful Check
928
foo Yes 00:05:00 2019-02-03T00:00:00
929
barbar Yes 00:05:00 2019-02-04T00:00:00
931
self.assertEqual(output, expected_output)
932
def test_verbose(self):
933
output = PrintTableCmd(verbose=True).output(
934
self.clients.values())
935
expected_output = """
936
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
937
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
938
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
940
self.assertEqual(output, expected_output)
941
def test_one_client(self):
942
output = PrintTableCmd().output(self.one_client.values())
943
expected_output = """
944
Name Enabled Timeout Last Successful Check
945
foo Yes 00:05:00 2019-02-03T00:00:00
947
self.assertEqual(output, expected_output)
949
class TestDumpJSONCmd(TestCmd):
951
self.expected_json = {
1999
954
"KeyID": ("92ed150794387c03ce684574b1139a65"
2000
955
"94a34f895daaaf09fd8ea90a27cddb12"),
2001
"Secret": b"secret",
2002
956
"Host": "foo.example.org",
2003
957
"Enabled": True,
2004
958
"Timeout": 300000,
2038
991
"ApprovedByDefault": False,
2039
992
"LastApprovalRequest": "2019-01-03T00:00:00",
2040
993
"ApprovalDelay": 30000,
2041
"ApprovalDuration": 93785000,
994
"ApprovalDuration": 1000,
2043
996
"ExtendedTimeout": 900000,
2044
997
"Expires": "2019-02-05T00:00:00",
2045
998
"LastCheckerStatus": -2,
2047
self.clients = collections.OrderedDict(
2049
("client_objectpath", self.client_properties),
2050
("other_client_objectpath",
2051
self.other_client_properties),
2053
self.one_client = {"client_objectpath":
2054
self.client_properties}
2055
self.testcase = testcase
2058
def call_method(self, methodname, busname, objectpath,
2060
self.testcase.assertEqual("se.recompile.Mandos", busname)
2061
self.calls.append((methodname, busname, objectpath,
2063
if interface == "org.freedesktop.DBus.Properties":
2064
if methodname == "Set":
2065
self.testcase.assertEqual(3, len(args))
2066
interface, key, value = args
2067
self.testcase.assertEqual(
2068
"se.recompile.Mandos.Client", interface)
2069
self.clients[objectpath][key] = value
2071
elif interface == "se.recompile.Mandos":
2072
self.testcase.assertEqual("RemoveClient", methodname)
2073
self.testcase.assertEqual(1, len(args))
2074
clientpath = args[0]
2075
del self.clients[clientpath]
2077
elif interface == "se.recompile.Mandos.Client":
2078
if methodname == "Approve":
2079
self.testcase.assertEqual(1, len(args))
2084
self.bus = self.FakeMandosBus(self)
2087
class TestBaseCommands(TestCommand):
2089
def test_IsEnabled_exits_successfully(self):
1001
return super(TestDumpJSONCmd, self).setUp()
1002
def test_normal(self):
1003
json_data = json.loads(DumpJSONCmd().output(self.clients))
1004
self.assertDictEqual(json_data, self.expected_json)
1005
def test_one_client(self):
1006
clients = self.one_client
1007
json_data = json.loads(DumpJSONCmd().output(clients))
1008
expected_json = {"foo": self.expected_json["foo"]}
1009
self.assertDictEqual(json_data, expected_json)
1011
class TestIsEnabledCmd(TestCmd):
1012
def test_is_enabled(self):
1013
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
1014
for client, properties in self.clients.items()))
1015
def test_is_enabled_run_exits_successfully(self):
2090
1016
with self.assertRaises(SystemExit) as e:
2091
command.IsEnabled().run(self.bus.one_client)
1017
IsEnabledCmd().run(self.one_client)
2092
1018
if e.exception.code is not None:
2093
self.assertEqual(0, e.exception.code)
1019
self.assertEqual(e.exception.code, 0)
2095
1021
self.assertIsNone(e.exception.code)
2097
def test_IsEnabled_exits_with_failure(self):
2098
self.bus.client_properties["Enabled"] = False
1022
def test_is_enabled_run_exits_with_failure(self):
1023
self.client.attributes["Enabled"] = dbus.Boolean(False)
2099
1024
with self.assertRaises(SystemExit) as e:
2100
command.IsEnabled().run(self.bus.one_client)
1025
IsEnabledCmd().run(self.one_client)
2101
1026
if isinstance(e.exception.code, int):
2102
self.assertNotEqual(0, e.exception.code)
1027
self.assertNotEqual(e.exception.code, 0)
2104
1029
self.assertIsNotNone(e.exception.code)
2106
def test_Approve(self):
2107
busname = "se.recompile.Mandos"
2108
client_interface = "se.recompile.Mandos.Client"
2109
command.Approve().run(self.bus.clients, self.bus)
2110
for clientpath in self.bus.clients:
2111
self.assertIn(("Approve", busname, clientpath,
2112
client_interface, (True,)), self.bus.calls)
2114
def test_Deny(self):
2115
busname = "se.recompile.Mandos"
2116
client_interface = "se.recompile.Mandos.Client"
2117
command.Deny().run(self.bus.clients, self.bus)
2118
for clientpath in self.bus.clients:
2119
self.assertIn(("Approve", busname, clientpath,
2120
client_interface, (False,)),
2123
def test_Remove(self):
2124
command.Remove().run(self.bus.clients, self.bus)
2125
for clientpath in self.bus.clients:
2126
self.assertIn(("RemoveClient", dbus_busname,
2127
dbus_server_path, dbus_server_interface,
2128
(clientpath,)), self.bus.calls)
2133
"KeyID": ("92ed150794387c03ce684574b1139a65"
2134
"94a34f895daaaf09fd8ea90a27cddb12"),
2135
"Host": "foo.example.org",
2138
"LastCheckedOK": "2019-02-03T00:00:00",
2139
"Created": "2019-01-02T00:00:00",
2141
"Fingerprint": ("778827225BA7DE539C5A"
2142
"7CFA59CFF7CDBD9A5920"),
2143
"CheckerRunning": False,
2144
"LastEnabled": "2019-01-03T00:00:00",
2145
"ApprovalPending": False,
2146
"ApprovedByDefault": True,
2147
"LastApprovalRequest": "",
2149
"ApprovalDuration": 1000,
2150
"Checker": "fping -q -- %(host)s",
2151
"ExtendedTimeout": 900000,
2152
"Expires": "2019-02-04T00:00:00",
2153
"LastCheckerStatus": 0,
2157
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
2158
"6ab612cff5ad227247e46c2b020f441c"),
2159
"Host": "192.0.2.3",
2162
"LastCheckedOK": "2019-02-04T00:00:00",
2163
"Created": "2019-01-03T00:00:00",
2165
"Fingerprint": ("3E393AEAEFB84C7E89E2"
2166
"F547B3A107558FCA3A27"),
2167
"CheckerRunning": True,
2168
"LastEnabled": "2019-01-04T00:00:00",
2169
"ApprovalPending": False,
2170
"ApprovedByDefault": False,
2171
"LastApprovalRequest": "2019-01-03T00:00:00",
2172
"ApprovalDelay": 30000,
2173
"ApprovalDuration": 93785000,
2175
"ExtendedTimeout": 900000,
2176
"Expires": "2019-02-05T00:00:00",
2177
"LastCheckerStatus": -2,
2181
def test_DumpJSON_normal(self):
2182
with self.capture_stdout_to_buffer() as buffer:
2183
command.DumpJSON().run(self.bus.clients)
2184
json_data = json.loads(buffer.getvalue())
2185
self.assertDictEqual(self.expected_json, json_data)
2188
@contextlib.contextmanager
2189
def capture_stdout_to_buffer():
2190
capture_buffer = io.StringIO()
2191
old_stdout = sys.stdout
2192
sys.stdout = capture_buffer
2194
yield capture_buffer
2196
sys.stdout = old_stdout
2198
def test_DumpJSON_one_client(self):
2199
with self.capture_stdout_to_buffer() as buffer:
2200
command.DumpJSON().run(self.bus.one_client)
2201
json_data = json.loads(buffer.getvalue())
2202
expected_json = {"foo": self.expected_json["foo"]}
2203
self.assertDictEqual(expected_json, json_data)
2205
def test_PrintTable_normal(self):
2206
with self.capture_stdout_to_buffer() as buffer:
2207
command.PrintTable().run(self.bus.clients)
2208
expected_output = "\n".join((
2209
"Name Enabled Timeout Last Successful Check",
2210
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2211
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
2213
self.assertEqual(expected_output, buffer.getvalue())
2215
def test_PrintTable_verbose(self):
2216
with self.capture_stdout_to_buffer() as buffer:
2217
command.PrintTable(verbose=True).run(self.bus.clients)
2232
"Last Successful Check ",
2233
"2019-02-03T00:00:00 ",
2234
"2019-02-04T00:00:00 ",
2237
"2019-01-02T00:00:00 ",
2238
"2019-01-03T00:00:00 ",
2250
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
2252
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
2256
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
2257
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
2259
"Check Is Running ",
2264
"2019-01-03T00:00:00 ",
2265
"2019-01-04T00:00:00 ",
2267
"Approval Is Pending ",
2271
"Approved By Default ",
2275
"Last Approval Request ",
2277
"2019-01-03T00:00:00 ",
2283
"Approval Duration ",
2288
"fping -q -- %(host)s ",
2291
"Extended Timeout ",
2296
"2019-02-04T00:00:00 ",
2297
"2019-02-05T00:00:00 ",
2299
"Last Checker Status",
2304
num_lines = max(len(rows) for rows in columns)
2305
expected_output = ("\n".join("".join(rows[line]
2306
for rows in columns)
2307
for line in range(num_lines))
2309
self.assertEqual(expected_output, buffer.getvalue())
2311
def test_PrintTable_one_client(self):
2312
with self.capture_stdout_to_buffer() as buffer:
2313
command.PrintTable().run(self.bus.one_client)
2314
expected_output = "\n".join((
2315
"Name Enabled Timeout Last Successful Check",
2316
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2318
self.assertEqual(expected_output, buffer.getvalue())
2321
class TestPropertySetterCmd(TestCommand):
2322
"""Abstract class for tests of command.PropertySetter classes"""
1031
class TestRemoveCmd(TestCmd):
1032
def test_remove(self):
1033
class MockMandos(object):
1036
def RemoveClient(self, dbus_path):
1037
self.calls.append(("RemoveClient", (dbus_path,)))
1038
mandos = MockMandos()
1039
super(TestRemoveCmd, self).setUp()
1040
RemoveCmd().run(self.clients, self.bus, mandos)
1041
self.assertEqual(len(mandos.calls), 2)
1042
for clientpath in self.clients:
1043
self.assertIn(("RemoveClient", (clientpath,)),
1046
class TestApproveCmd(TestCmd):
1047
def test_approve(self):
1048
ApproveCmd().run(self.clients, self.bus)
1049
for clientpath in self.clients:
1050
client = self.bus.get_object(busname, clientpath)
1051
self.assertIn(("Approve", (True, client_interface)),
1054
class TestDenyCmd(TestCmd):
1055
def test_deny(self):
1056
DenyCmd().run(self.clients, self.bus)
1057
for clientpath in self.clients:
1058
client = self.bus.get_object(busname, clientpath)
1059
self.assertIn(("Approve", (False, client_interface)),
1062
class TestEnableCmd(TestCmd):
1063
def test_enable(self):
1064
for clientpath in self.clients:
1065
client = self.bus.get_object(busname, clientpath)
1066
client.attributes["Enabled"] = False
1068
EnableCmd().run(self.clients, self.bus)
1070
for clientpath in self.clients:
1071
client = self.bus.get_object(busname, clientpath)
1072
self.assertTrue(client.attributes["Enabled"])
1074
class TestDisableCmd(TestCmd):
1075
def test_disable(self):
1076
DisableCmd().run(self.clients, self.bus)
1077
for clientpath in self.clients:
1078
client = self.bus.get_object(busname, clientpath)
1079
self.assertFalse(client.attributes["Enabled"])
1081
class Unique(object):
1082
"""Class for objects which exist only to be unique objects, since
1083
unittest.mock.sentinel only exists in Python 3.3"""
1085
class TestPropertyCmd(TestCmd):
1086
"""Abstract class for tests of PropertyCmd classes"""
2324
1087
def runTest(self):
2325
1088
if not hasattr(self, "command"):
2326
return # Abstract TestCase class
2328
if hasattr(self, "values_to_set"):
2329
cmd_args = [(value,) for value in self.values_to_set]
2330
values_to_get = getattr(self, "values_to_get",
2333
cmd_args = [() for x in range(len(self.values_to_get))]
2334
values_to_get = self.values_to_get
2335
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2336
for clientpath in self.bus.clients:
2337
self.bus.clients[clientpath][self.propname] = (
2339
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2340
for clientpath in self.bus.clients:
2341
value = (self.bus.clients[clientpath]
1090
values_to_get = getattr(self, "values_to_get",
1092
for value_to_set, value_to_get in zip(self.values_to_set,
1094
for clientpath in self.clients:
1095
client = self.bus.get_object(busname, clientpath)
1096
old_value = client.attributes[self.propname]
1097
self.assertNotIsInstance(old_value, Unique)
1098
client.attributes[self.propname] = Unique()
1099
self.run_command(value_to_set, self.clients)
1100
for clientpath in self.clients:
1101
client = self.bus.get_object(busname, clientpath)
1102
value = client.attributes[self.propname]
2343
1103
self.assertNotIsInstance(value, Unique)
2344
self.assertEqual(value_to_get, value)
2347
class TestEnableCmd(TestPropertySetterCmd):
2348
command = command.Enable
2349
propname = "Enabled"
2350
values_to_get = [True]
2353
class TestDisableCmd(TestPropertySetterCmd):
2354
command = command.Disable
2355
propname = "Enabled"
2356
values_to_get = [False]
2359
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2360
command = command.BumpTimeout
1104
self.assertEqual(value, value_to_get)
1105
def run_command(self, value, clients):
1106
self.command().run(clients, self.bus)
1108
class TestBumpTimeoutCmd(TestPropertyCmd):
1109
command = BumpTimeoutCmd
2361
1110
propname = "LastCheckedOK"
2362
values_to_get = [""]
2365
class TestStartCheckerCmd(TestPropertySetterCmd):
2366
command = command.StartChecker
2367
propname = "CheckerRunning"
2368
values_to_get = [True]
2371
class TestStopCheckerCmd(TestPropertySetterCmd):
2372
command = command.StopChecker
2373
propname = "CheckerRunning"
2374
values_to_get = [False]
2377
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2378
command = command.ApproveByDefault
2379
propname = "ApprovedByDefault"
2380
values_to_get = [True]
2383
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2384
command = command.DenyByDefault
2385
propname = "ApprovedByDefault"
2386
values_to_get = [False]
2389
class TestSetCheckerCmd(TestPropertySetterCmd):
2390
command = command.SetChecker
1111
values_to_set = [""]
1113
class TestStartCheckerCmd(TestPropertyCmd):
1114
command = StartCheckerCmd
1115
propname = "CheckerRunning"
1116
values_to_set = [dbus.Boolean(True)]
1118
class TestStopCheckerCmd(TestPropertyCmd):
1119
command = StopCheckerCmd
1120
propname = "CheckerRunning"
1121
values_to_set = [dbus.Boolean(False)]
1123
class TestApproveByDefaultCmd(TestPropertyCmd):
1124
command = ApproveByDefaultCmd
1125
propname = "ApprovedByDefault"
1126
values_to_set = [dbus.Boolean(True)]
1128
class TestDenyByDefaultCmd(TestPropertyCmd):
1129
command = DenyByDefaultCmd
1130
propname = "ApprovedByDefault"
1131
values_to_set = [dbus.Boolean(False)]
1133
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1134
"""Abstract class for tests of PropertyCmd classes using the
1135
ValueArgumentMixIn"""
1137
if type(self) is TestValueArgumentPropertyCmd:
1139
return super(TestValueArgumentPropertyCmd, self).runTest()
1140
def run_command(self, value, clients):
1141
self.command(value).run(clients, self.bus)
1143
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1144
command = SetCheckerCmd
2391
1145
propname = "Checker"
2392
1146
values_to_set = ["", ":", "fping -q -- %s"]
2395
class TestSetHostCmd(TestPropertySetterCmd):
2396
command = command.SetHost
1148
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1149
command = SetHostCmd
2397
1150
propname = "Host"
2398
values_to_set = ["192.0.2.3", "client.example.org"]
2401
class TestSetSecretCmd(TestPropertySetterCmd):
2402
command = command.SetSecret
1151
values_to_set = ["192.0.2.3", "foo.example.org"]
1153
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1154
command = SetSecretCmd
2403
1155
propname = "Secret"
2404
1156
values_to_set = [io.BytesIO(b""),
2405
1157
io.BytesIO(b"secret\0xyzzy\nbar")]
2406
values_to_get = [f.getvalue() for f in values_to_set]
2409
class TestSetTimeoutCmd(TestPropertySetterCmd):
2410
command = command.SetTimeout
1158
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1160
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1161
command = SetTimeoutCmd
2411
1162
propname = "Timeout"
2412
1163
values_to_set = [datetime.timedelta(),
2413
1164
datetime.timedelta(minutes=5),
2414
1165
datetime.timedelta(seconds=1),
2415
1166
datetime.timedelta(weeks=1),
2416
1167
datetime.timedelta(weeks=52)]
2417
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2420
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
2421
command = command.SetExtendedTimeout
1168
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1170
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1171
command = SetExtendedTimeoutCmd
2422
1172
propname = "ExtendedTimeout"
2423
1173
values_to_set = [datetime.timedelta(),
2424
1174
datetime.timedelta(minutes=5),
2425
1175
datetime.timedelta(seconds=1),
2426
1176
datetime.timedelta(weeks=1),
2427
1177
datetime.timedelta(weeks=52)]
2428
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2431
class TestSetIntervalCmd(TestPropertySetterCmd):
2432
command = command.SetInterval
1178
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1180
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1181
command = SetIntervalCmd
2433
1182
propname = "Interval"
2434
1183
values_to_set = [datetime.timedelta(),
2435
1184
datetime.timedelta(minutes=5),
2436
1185
datetime.timedelta(seconds=1),
2437
1186
datetime.timedelta(weeks=1),
2438
1187
datetime.timedelta(weeks=52)]
2439
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2442
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
2443
command = command.SetApprovalDelay
1188
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1190
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1191
command = SetApprovalDelayCmd
2444
1192
propname = "ApprovalDelay"
2445
1193
values_to_set = [datetime.timedelta(),
2446
1194
datetime.timedelta(minutes=5),
2447
1195
datetime.timedelta(seconds=1),
2448
1196
datetime.timedelta(weeks=1),
2449
1197
datetime.timedelta(weeks=52)]
2450
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2453
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
2454
command = command.SetApprovalDuration
1198
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1200
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1201
command = SetApprovalDurationCmd
2455
1202
propname = "ApprovalDuration"
2456
1203
values_to_set = [datetime.timedelta(),
2457
1204
datetime.timedelta(minutes=5),
2458
1205
datetime.timedelta(seconds=1),
2459
1206
datetime.timedelta(weeks=1),
2460
1207
datetime.timedelta(weeks=52)]
2461
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1208
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1210
class Test_command_from_options(unittest.TestCase):
1212
self.parser = argparse.ArgumentParser()
1213
add_command_line_options(self.parser)
1214
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1215
"""Assert that parsing ARGS should result in an instance of
1216
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1217
options = self.parser.parse_args(args)
1218
check_option_syntax(self.parser, options)
1219
commands = commands_from_options(options)
1220
self.assertEqual(len(commands), 1)
1221
command = commands[0]
1222
self.assertIsInstance(command, command_cls)
1223
for key, value in cmd_attrs.items():
1224
self.assertEqual(getattr(command, key), value)
1225
def test_print_table(self):
1226
self.assert_command_from_args([], PrintTableCmd,
1229
def test_print_table_verbose(self):
1230
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1233
def test_print_table_verbose_short(self):
1234
self.assert_command_from_args(["-v"], PrintTableCmd,
1237
def test_enable(self):
1238
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1240
def test_enable_short(self):
1241
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1243
def test_disable(self):
1244
self.assert_command_from_args(["--disable", "foo"],
1247
def test_disable_short(self):
1248
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1250
def test_bump_timeout(self):
1251
self.assert_command_from_args(["--bump-timeout", "foo"],
1254
def test_bump_timeout_short(self):
1255
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1257
def test_start_checker(self):
1258
self.assert_command_from_args(["--start-checker", "foo"],
1261
def test_stop_checker(self):
1262
self.assert_command_from_args(["--stop-checker", "foo"],
1265
def test_remove(self):
1266
self.assert_command_from_args(["--remove", "foo"],
1269
def test_remove_short(self):
1270
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1272
def test_checker(self):
1273
self.assert_command_from_args(["--checker", ":", "foo"],
1274
SetCheckerCmd, value_to_set=":")
1276
def test_checker_empty(self):
1277
self.assert_command_from_args(["--checker", "", "foo"],
1278
SetCheckerCmd, value_to_set="")
1280
def test_checker_short(self):
1281
self.assert_command_from_args(["-c", ":", "foo"],
1282
SetCheckerCmd, value_to_set=":")
1284
def test_timeout(self):
1285
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1287
value_to_set=300000)
1289
def test_timeout_short(self):
1290
self.assert_command_from_args(["-t", "PT5M", "foo"],
1292
value_to_set=300000)
1294
def test_extended_timeout(self):
1295
self.assert_command_from_args(["--extended-timeout", "PT15M",
1297
SetExtendedTimeoutCmd,
1298
value_to_set=900000)
1300
def test_interval(self):
1301
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1303
value_to_set=120000)
1305
def test_interval_short(self):
1306
self.assert_command_from_args(["-i", "PT2M", "foo"],
1308
value_to_set=120000)
1310
def test_approve_by_default(self):
1311
self.assert_command_from_args(["--approve-by-default", "foo"],
1312
ApproveByDefaultCmd)
1314
def test_deny_by_default(self):
1315
self.assert_command_from_args(["--deny-by-default", "foo"],
1318
def test_approval_delay(self):
1319
self.assert_command_from_args(["--approval-delay", "PT30S",
1320
"foo"], SetApprovalDelayCmd,
1323
def test_approval_duration(self):
1324
self.assert_command_from_args(["--approval-duration", "PT1S",
1325
"foo"], SetApprovalDurationCmd,
1328
def test_host(self):
1329
self.assert_command_from_args(["--host", "foo.example.org",
1331
value_to_set="foo.example.org")
1333
def test_host_short(self):
1334
self.assert_command_from_args(["-H", "foo.example.org",
1336
value_to_set="foo.example.org")
1338
def test_secret_devnull(self):
1339
self.assert_command_from_args(["--secret", os.path.devnull,
1340
"foo"], SetSecretCmd,
1343
def test_secret_tempfile(self):
1344
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1345
value = b"secret\0xyzzy\nbar"
1348
self.assert_command_from_args(["--secret", f.name,
1349
"foo"], SetSecretCmd,
1352
def test_secret_devnull_short(self):
1353
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1354
SetSecretCmd, value_to_set=b"")
1356
def test_secret_tempfile_short(self):
1357
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1358
value = b"secret\0xyzzy\nbar"
1361
self.assert_command_from_args(["-s", f.name, "foo"],
1365
def test_approve(self):
1366
self.assert_command_from_args(["--approve", "foo"],
1369
def test_approve_short(self):
1370
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1372
def test_deny(self):
1373
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1375
def test_deny_short(self):
1376
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1378
def test_dump_json(self):
1379
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1381
def test_is_enabled(self):
1382
self.assert_command_from_args(["--is-enabled", "foo"],
1385
def test_is_enabled_short(self):
1386
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1388
def test_deny_before_remove(self):
1389
options = self.parser.parse_args(["--deny", "--remove", "foo"])
1390
check_option_syntax(self.parser, options)
1391
commands = commands_from_options(options)
1392
self.assertEqual(len(commands), 2)
1393
self.assertIsInstance(commands[0], DenyCmd)
1394
self.assertIsInstance(commands[1], RemoveCmd)
1396
def test_deny_before_remove_reversed(self):
1397
options = self.parser.parse_args(["--remove", "--deny", "--all"])
1398
check_option_syntax(self.parser, options)
1399
commands = commands_from_options(options)
1400
self.assertEqual(len(commands), 2)
1401
self.assertIsInstance(commands[0], DenyCmd)
1402
self.assertIsInstance(commands[1], RemoveCmd)
1405
class Test_check_option_syntax(unittest.TestCase):
1406
# This mostly corresponds to the definition from has_actions() in
1407
# check_option_syntax()
1409
# The actual values set here are not that important, but we do
1410
# at least stick to the correct types, even though they are
1414
"bump_timeout": True,
1415
"start_checker": True,
1416
"stop_checker": True,
1420
"timeout": datetime.timedelta(),
1421
"extended_timeout": datetime.timedelta(),
1422
"interval": datetime.timedelta(),
1423
"approved_by_default": True,
1424
"approval_delay": datetime.timedelta(),
1425
"approval_duration": datetime.timedelta(),
1427
"secret": io.BytesIO(b"x"),
1433
self.parser = argparse.ArgumentParser()
1434
add_command_line_options(self.parser)
1436
@contextlib.contextmanager
1437
def assertParseError(self):
1438
with self.assertRaises(SystemExit) as e:
1439
with self.temporarily_suppress_stderr():
1441
# Exit code from argparse is guaranteed to be "2". Reference:
1442
# https://docs.python.org/3/library/argparse.html#exiting-methods
1443
self.assertEqual(e.exception.code, 2)
1446
@contextlib.contextmanager
1447
def temporarily_suppress_stderr():
1448
null = os.open(os.path.devnull, os.O_RDWR)
1449
stderrcopy = os.dup(sys.stderr.fileno())
1450
os.dup2(null, sys.stderr.fileno())
1456
os.dup2(stderrcopy, sys.stderr.fileno())
1457
os.close(stderrcopy)
1459
def check_option_syntax(self, options):
1460
check_option_syntax(self.parser, options)
1462
def test_actions_requires_client_or_all(self):
1463
for action, value in self.actions.items():
1464
options = self.parser.parse_args()
1465
setattr(options, action, value)
1466
with self.assertParseError():
1467
self.check_option_syntax(options)
1469
def test_actions_conflicts_with_verbose(self):
1470
for action, value in self.actions.items():
1471
options = self.parser.parse_args()
1472
setattr(options, action, value)
1473
options.verbose = True
1474
with self.assertParseError():
1475
self.check_option_syntax(options)
1477
def test_dump_json_conflicts_with_verbose(self):
1478
options = self.parser.parse_args()
1479
options.dump_json = True
1480
options.verbose = True
1481
with self.assertParseError():
1482
self.check_option_syntax(options)
1484
def test_dump_json_conflicts_with_action(self):
1485
for action, value in self.actions.items():
1486
options = self.parser.parse_args()
1487
setattr(options, action, value)
1488
options.dump_json = True
1489
with self.assertParseError():
1490
self.check_option_syntax(options)
1492
def test_all_can_not_be_alone(self):
1493
options = self.parser.parse_args()
1495
with self.assertParseError():
1496
self.check_option_syntax(options)
1498
def test_all_is_ok_with_any_action(self):
1499
for action, value in self.actions.items():
1500
options = self.parser.parse_args()
1501
setattr(options, action, value)
1503
self.check_option_syntax(options)
1505
def test_is_enabled_fails_without_client(self):
1506
options = self.parser.parse_args()
1507
options.is_enabled = True
1508
with self.assertParseError():
1509
self.check_option_syntax(options)
1511
def test_is_enabled_works_with_one_client(self):
1512
options = self.parser.parse_args()
1513
options.is_enabled = True
1514
options.client = ["foo"]
1515
self.check_option_syntax(options)
1517
def test_is_enabled_fails_with_two_clients(self):
1518
options = self.parser.parse_args()
1519
options.is_enabled = True
1520
options.client = ["foo", "barbar"]
1521
with self.assertParseError():
1522
self.check_option_syntax(options)
1524
def test_remove_can_only_be_combined_with_action_deny(self):
1525
for action, value in self.actions.items():
1526
if action in {"remove", "deny"}:
1528
options = self.parser.parse_args()
1529
setattr(options, action, value)
1531
options.remove = True
1532
with self.assertParseError():
1533
self.check_option_syntax(options)