274
## Classes for commands.
276
# Abstract classes first
277
class Command(object):
278
"""Abstract class for commands"""
279
def run(self, mandos, clients):
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 client, properties in clients.items():
285
self.run_on_one_client(client, properties)
287
class PrintCmd(Command):
288
"""Abstract class for commands printing client details"""
289
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
290
"Created", "Interval", "Host", "KeyID",
291
"Fingerprint", "CheckerRunning", "LastEnabled",
292
"ApprovalPending", "ApprovedByDefault",
293
"LastApprovalRequest", "ApprovalDelay",
294
"ApprovalDuration", "Checker", "ExtendedTimeout",
295
"Expires", "LastCheckerStatus")
296
def run(self, mandos, clients):
297
print(self.output(clients))
299
class PropertyCmd(Command):
300
"""Abstract class for Actions for setting one client property"""
301
def run_on_one_client(self, client, properties):
302
"""Set the Client's D-Bus property"""
303
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
304
client.__dbus_object_path__,
305
dbus.PROPERTIES_IFACE, client_interface,
306
self.property, self.value_to_set
307
if not isinstance(self.value_to_set, dbus.Boolean)
308
else bool(self.value_to_set))
309
client.Set(client_interface, self.property, self.value_to_set,
310
dbus_interface=dbus.PROPERTIES_IFACE)
312
class ValueArgumentMixIn(object):
313
"""Mixin class for commands taking a value as argument"""
314
def __init__(self, value):
315
self.value_to_set = value
317
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
318
"""Mixin class for commands taking a value argument as
321
def value_to_set(self):
324
def value_to_set(self, value):
325
"""When setting, convert value to a datetime.timedelta"""
326
self._vts = int(round(value.total_seconds() * 1000))
328
# Actual (non-abstract) command classes
330
class PrintTableCmd(PrintCmd):
331
def __init__(self, verbose=False):
332
self.verbose = verbose
334
def output(self, clients):
335
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
336
keywords = default_keywords
338
keywords = self.all_keywords
339
return str(self.TableOfClients(clients.values(), keywords))
341
class TableOfClients(object):
344
"Enabled": "Enabled",
345
"Timeout": "Timeout",
346
"LastCheckedOK": "Last Successful Check",
347
"LastApprovalRequest": "Last Approval Request",
348
"Created": "Created",
349
"Interval": "Interval",
351
"Fingerprint": "Fingerprint",
353
"CheckerRunning": "Check Is Running",
354
"LastEnabled": "Last Enabled",
355
"ApprovalPending": "Approval Is Pending",
356
"ApprovedByDefault": "Approved By Default",
357
"ApprovalDelay": "Approval Delay",
358
"ApprovalDuration": "Approval Duration",
359
"Checker": "Checker",
360
"ExtendedTimeout": "Extended Timeout",
361
"Expires": "Expires",
362
"LastCheckerStatus": "Last Checker Status",
365
def __init__(self, clients, keywords, tableheaders=None):
366
self.clients = clients
367
self.keywords = keywords
368
if tableheaders is not None:
369
self.tableheaders = tableheaders
372
return "\n".join(self.rows())
374
if sys.version_info.major == 2:
375
__unicode__ = __str__
377
return str(self).encode(locale.getpreferredencoding())
380
format_string = self.row_formatting_string()
381
rows = [self.header_line(format_string)]
382
rows.extend(self.client_line(client, format_string)
383
for client in self.clients)
386
def row_formatting_string(self):
387
"Format string used to format table rows"
388
return " ".join("{{{key}:{width}}}".format(
389
width=max(len(self.tableheaders[key]),
390
*(len(self.string_from_client(client, key))
391
for client in self.clients)),
393
for key in self.keywords)
395
def string_from_client(self, client, key):
396
return self.valuetostring(client[key], key)
399
def valuetostring(value, keyword):
400
if isinstance(value, dbus.Boolean):
401
return "Yes" if value else "No"
402
if keyword in ("Timeout", "Interval", "ApprovalDelay",
403
"ApprovalDuration", "ExtendedTimeout"):
404
return milliseconds_to_string(value)
407
def header_line(self, format_string):
408
return format_string.format(**self.tableheaders)
410
def client_line(self, client, format_string):
411
return format_string.format(
412
**{key: self.string_from_client(client, key)
413
for key in self.keywords})
417
class DumpJSONCmd(PrintCmd):
418
def output(self, clients):
419
data = {client["Name"]:
420
{key: self.dbus_boolean_to_bool(client[key])
421
for key in self.all_keywords}
422
for client in clients.values()}
423
return json.dumps(data, indent=4, separators=(',', ': '))
425
def dbus_boolean_to_bool(value):
426
if isinstance(value, dbus.Boolean):
430
class IsEnabledCmd(Command):
431
def run_on_one_client(self, client, properties):
432
if self.is_enabled(client, properties):
435
def is_enabled(self, client, properties):
436
return bool(properties["Enabled"])
438
class RemoveCmd(Command):
439
def run_on_one_client(self, client, properties):
440
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", busname,
441
server_path, server_interface,
442
str(client.__dbus_object_path__))
443
self.mandos.RemoveClient(client.__dbus_object_path__)
445
class ApproveCmd(Command):
446
def run_on_one_client(self, client, properties):
447
log.debug("D-Bus: %s:%s.Approve(True)",
448
client.__dbus_object_path__, client_interface)
449
client.Approve(dbus.Boolean(True),
450
dbus_interface=client_interface)
452
class DenyCmd(Command):
453
def run_on_one_client(self, client, properties):
454
log.debug("D-Bus: %s:%s.Approve(False)",
455
client.__dbus_object_path__, client_interface)
456
client.Approve(dbus.Boolean(False),
457
dbus_interface=client_interface)
459
class EnableCmd(PropertyCmd):
461
value_to_set = dbus.Boolean(True)
463
class DisableCmd(PropertyCmd):
465
value_to_set = dbus.Boolean(False)
467
class BumpTimeoutCmd(PropertyCmd):
468
property = "LastCheckedOK"
471
class StartCheckerCmd(PropertyCmd):
472
property = "CheckerRunning"
473
value_to_set = dbus.Boolean(True)
475
class StopCheckerCmd(PropertyCmd):
476
property = "CheckerRunning"
477
value_to_set = dbus.Boolean(False)
479
class ApproveByDefaultCmd(PropertyCmd):
480
property = "ApprovedByDefault"
481
value_to_set = dbus.Boolean(True)
483
class DenyByDefaultCmd(PropertyCmd):
484
property = "ApprovedByDefault"
485
value_to_set = dbus.Boolean(False)
487
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
490
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
493
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
495
def value_to_set(self):
498
def value_to_set(self, value):
499
"""When setting, read data from supplied file object"""
500
self._vts = value.read()
504
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
507
class SetExtendedTimeoutCmd(PropertyCmd,
508
MillisecondsValueArgumentMixIn):
509
property = "ExtendedTimeout"
511
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
512
property = "Interval"
514
class SetApprovalDelayCmd(PropertyCmd,
515
MillisecondsValueArgumentMixIn):
516
property = "ApprovalDelay"
518
class SetApprovalDurationCmd(PropertyCmd,
519
MillisecondsValueArgumentMixIn):
520
property = "ApprovalDuration"
522
def add_command_line_options(parser):
523
parser.add_argument("--version", action="version",
524
version="%(prog)s {}".format(version),
525
help="show version number and exit")
526
parser.add_argument("-a", "--all", action="store_true",
527
help="Select all clients")
528
parser.add_argument("-v", "--verbose", action="store_true",
529
help="Print all fields")
530
parser.add_argument("-j", "--dump-json", action="store_true",
531
help="Dump client data in JSON format")
532
enable_disable = parser.add_mutually_exclusive_group()
533
enable_disable.add_argument("-e", "--enable", action="store_true",
534
help="Enable client")
535
enable_disable.add_argument("-d", "--disable",
537
help="disable client")
538
parser.add_argument("-b", "--bump-timeout", action="store_true",
539
help="Bump timeout for client")
540
start_stop_checker = parser.add_mutually_exclusive_group()
541
start_stop_checker.add_argument("--start-checker",
543
help="Start checker for client")
544
start_stop_checker.add_argument("--stop-checker",
546
help="Stop checker for client")
547
parser.add_argument("-V", "--is-enabled", action="store_true",
548
help="Check if client is enabled")
549
parser.add_argument("-r", "--remove", action="store_true",
550
help="Remove client")
551
parser.add_argument("-c", "--checker",
552
help="Set checker command for client")
553
parser.add_argument("-t", "--timeout", type=string_to_delta,
554
help="Set timeout for client")
555
parser.add_argument("--extended-timeout", type=string_to_delta,
556
help="Set extended timeout for client")
557
parser.add_argument("-i", "--interval", type=string_to_delta,
558
help="Set checker interval for client")
559
approve_deny_default = parser.add_mutually_exclusive_group()
560
approve_deny_default.add_argument(
561
"--approve-by-default", action="store_true",
562
default=None, dest="approved_by_default",
563
help="Set client to be approved by default")
564
approve_deny_default.add_argument(
565
"--deny-by-default", action="store_false",
566
dest="approved_by_default",
567
help="Set client to be denied by default")
568
parser.add_argument("--approval-delay", type=string_to_delta,
569
help="Set delay before client approve/deny")
570
parser.add_argument("--approval-duration", type=string_to_delta,
571
help="Set duration of one client approval")
572
parser.add_argument("-H", "--host", help="Set host for client")
573
parser.add_argument("-s", "--secret",
574
type=argparse.FileType(mode="rb"),
575
help="Set password blob (file) for client")
576
approve_deny = parser.add_mutually_exclusive_group()
577
approve_deny.add_argument(
578
"-A", "--approve", action="store_true",
579
help="Approve any current client request")
580
approve_deny.add_argument("-D", "--deny", action="store_true",
581
help="Deny any current client request")
582
parser.add_argument("--debug", action="store_true",
583
help="Debug mode (show D-Bus commands)")
584
parser.add_argument("--check", action="store_true",
585
help="Run self-test")
586
parser.add_argument("client", nargs="*", help="Client name")
589
def commands_from_options(options):
593
if options.dump_json:
594
commands.append(DumpJSONCmd())
597
commands.append(EnableCmd())
600
commands.append(DisableCmd())
602
if options.bump_timeout:
603
commands.append(BumpTimeoutCmd())
605
if options.start_checker:
606
commands.append(StartCheckerCmd())
608
if options.stop_checker:
609
commands.append(StopCheckerCmd())
611
if options.is_enabled:
612
commands.append(IsEnabledCmd())
614
if options.checker is not None:
615
commands.append(SetCheckerCmd(options.checker))
617
if options.timeout is not None:
618
commands.append(SetTimeoutCmd(options.timeout))
620
if options.extended_timeout:
622
SetExtendedTimeoutCmd(options.extended_timeout))
624
if options.interval is not None:
625
commands.append(SetIntervalCmd(options.interval))
627
if options.approved_by_default is not None:
628
if options.approved_by_default:
629
commands.append(ApproveByDefaultCmd())
631
commands.append(DenyByDefaultCmd())
633
if options.approval_delay is not None:
634
commands.append(SetApprovalDelayCmd(options.approval_delay))
636
if options.approval_duration is not None:
638
SetApprovalDurationCmd(options.approval_duration))
640
if options.host is not None:
641
commands.append(SetHostCmd(options.host))
643
if options.secret is not None:
644
commands.append(SetSecretCmd(options.secret))
647
commands.append(ApproveCmd())
650
commands.append(DenyCmd())
653
commands.append(RemoveCmd())
655
# If no command option has been given, show table of clients,
656
# optionally verbosely
658
commands.append(PrintTableCmd(verbose=options.verbose))
385
663
def check_option_syntax(parser, options):
386
664
"""Apply additional restrictions on options, not expressible in
424
702
options.remove = True
427
def get_mandos_dbus_object(bus):
428
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
429
dbus_busname, server_dbus_path)
430
with if_dbus_exception_log_with_exception_and_exit(
431
"Could not connect to Mandos server: %s"):
432
mandos_dbus_object = bus.get_object(dbus_busname,
434
return mandos_dbus_object
437
@contextlib.contextmanager
438
def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):
706
parser = argparse.ArgumentParser()
708
add_command_line_options(parser)
710
options = parser.parse_args()
712
check_option_syntax(parser, options)
714
clientnames = options.client
717
log.setLevel(logging.DEBUG)
441
except dbus.exceptions.DBusException as e:
442
log.critical(*(args + (e,)), **kwargs)
720
bus = dbus.SystemBus()
721
log.debug("D-Bus: Connect to: (name=%r, path=%r)", busname,
723
mandos_dbus_objc = bus.get_object(busname, server_path)
724
except dbus.exceptions.DBusException:
725
log.critical("Could not connect to Mandos server")
446
def get_managed_objects(object_manager):
447
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
448
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
449
with if_dbus_exception_log_with_exception_and_exit(
450
"Failed to access Mandos server through D-Bus:\n%s"):
451
with SilenceLogger("dbus.proxies"):
452
managed_objects = object_manager.GetManagedObjects()
453
return managed_objects
456
class SilenceLogger(object):
457
"Simple context manager to silence a particular logger"
458
def __init__(self, loggername):
459
self.logger = logging.getLogger(loggername)
462
self.logger.addFilter(self.nullfilter)
728
mandos_serv = dbus.Interface(mandos_dbus_objc,
729
dbus_interface=server_interface)
730
mandos_serv_object_manager = dbus.Interface(
731
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
733
# Filter out log message from dbus module
734
dbus_logger = logging.getLogger("dbus.proxies")
465
735
class NullFilter(logging.Filter):
466
736
def filter(self, record):
469
nullfilter = NullFilter()
471
def __exit__(self, exc_type, exc_val, exc_tb):
472
self.logger.removeFilter(self.nullfilter)
475
def commands_from_options(options):
479
if options.is_enabled:
480
commands.append(command.IsEnabled())
483
commands.append(command.Approve())
486
commands.append(command.Deny())
489
commands.append(command.Remove())
491
if options.dump_json:
492
commands.append(command.DumpJSON())
495
commands.append(command.Enable())
498
commands.append(command.Disable())
500
if options.bump_timeout:
501
commands.append(command.BumpTimeout())
503
if options.start_checker:
504
commands.append(command.StartChecker())
506
if options.stop_checker:
507
commands.append(command.StopChecker())
509
if options.approved_by_default is not None:
510
if options.approved_by_default:
511
commands.append(command.ApproveByDefault())
513
commands.append(command.DenyByDefault())
515
if options.checker is not None:
516
commands.append(command.SetChecker(options.checker))
518
if options.host is not None:
519
commands.append(command.SetHost(options.host))
521
if options.secret is not None:
522
commands.append(command.SetSecret(options.secret))
524
if options.timeout is not None:
525
commands.append(command.SetTimeout(options.timeout))
527
if options.extended_timeout:
529
command.SetExtendedTimeout(options.extended_timeout))
531
if options.interval is not None:
532
commands.append(command.SetInterval(options.interval))
534
if options.approval_delay is not None:
536
command.SetApprovalDelay(options.approval_delay))
538
if options.approval_duration is not None:
540
command.SetApprovalDuration(options.approval_duration))
542
# If no command option has been given, show table of clients,
543
# optionally verbosely
545
commands.append(command.PrintTable(verbose=options.verbose))
550
class command(object):
551
"""A namespace for command classes"""
554
"""Abstract base class for commands"""
555
def run(self, clients, bus=None, mandos=None):
556
"""Normal commands should implement run_on_one_client(),
557
but commands which want to operate on all clients at the same time can
558
override this run() method instead.
561
for clientpath, properties in clients.items():
562
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
563
dbus_busname, str(clientpath))
564
client = bus.get_object(dbus_busname, clientpath)
565
self.run_on_one_client(client, properties)
568
class IsEnabled(Base):
569
def run(self, clients, bus=None, mandos=None):
570
client, properties = next(iter(clients.items()))
571
if self.is_enabled(client, properties):
574
def is_enabled(self, client, properties):
575
return properties["Enabled"]
579
def run_on_one_client(self, client, properties):
580
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
581
client.__dbus_object_path__,
582
client_dbus_interface)
583
client.Approve(dbus.Boolean(True),
584
dbus_interface=client_dbus_interface)
588
def run_on_one_client(self, client, properties):
589
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
590
client.__dbus_object_path__,
591
client_dbus_interface)
592
client.Approve(dbus.Boolean(False),
593
dbus_interface=client_dbus_interface)
597
def run_on_one_client(self, client, properties):
598
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
599
dbus_busname, server_dbus_path,
600
server_dbus_interface,
601
str(client.__dbus_object_path__))
602
self.mandos.RemoveClient(client.__dbus_object_path__)
606
"""Abstract class for commands outputting client details"""
607
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
608
"Created", "Interval", "Host", "KeyID",
609
"Fingerprint", "CheckerRunning",
610
"LastEnabled", "ApprovalPending",
611
"ApprovedByDefault", "LastApprovalRequest",
612
"ApprovalDelay", "ApprovalDuration",
613
"Checker", "ExtendedTimeout", "Expires",
616
def run(self, clients, bus=None, mandos=None):
617
print(self.output(clients.values()))
619
def output(self, clients):
620
raise NotImplementedError()
623
class DumpJSON(Output):
624
def output(self, clients):
625
data = {client["Name"]:
626
{key: self.dbus_boolean_to_bool(client[key])
627
for key in self.all_keywords}
628
for client in clients}
629
return json.dumps(data, indent=4, separators=(',', ': '))
632
def dbus_boolean_to_bool(value):
633
if isinstance(value, dbus.Boolean):
638
class PrintTable(Output):
639
def __init__(self, verbose=False):
640
self.verbose = verbose
642
def output(self, clients):
643
default_keywords = ("Name", "Enabled", "Timeout",
645
keywords = default_keywords
647
keywords = self.all_keywords
648
return str(self.TableOfClients(clients, keywords))
650
class TableOfClients(object):
653
"Enabled": "Enabled",
654
"Timeout": "Timeout",
655
"LastCheckedOK": "Last Successful Check",
656
"LastApprovalRequest": "Last Approval Request",
657
"Created": "Created",
658
"Interval": "Interval",
660
"Fingerprint": "Fingerprint",
662
"CheckerRunning": "Check Is Running",
663
"LastEnabled": "Last Enabled",
664
"ApprovalPending": "Approval Is Pending",
665
"ApprovedByDefault": "Approved By Default",
666
"ApprovalDelay": "Approval Delay",
667
"ApprovalDuration": "Approval Duration",
668
"Checker": "Checker",
669
"ExtendedTimeout": "Extended Timeout",
670
"Expires": "Expires",
671
"LastCheckerStatus": "Last Checker Status",
674
def __init__(self, clients, keywords):
675
self.clients = clients
676
self.keywords = keywords
679
return "\n".join(self.rows())
681
if sys.version_info.major == 2:
682
__unicode__ = __str__
684
return str(self).encode(
685
locale.getpreferredencoding())
688
format_string = self.row_formatting_string()
689
rows = [self.header_line(format_string)]
690
rows.extend(self.client_line(client, format_string)
691
for client in self.clients)
694
def row_formatting_string(self):
695
"Format string used to format table rows"
696
return " ".join("{{{key}:{width}}}".format(
697
width=max(len(self.tableheaders[key]),
698
*(len(self.string_from_client(client,
700
for client in self.clients)),
702
for key in self.keywords)
704
def string_from_client(self, client, key):
705
return self.valuetostring(client[key], key)
708
def valuetostring(cls, value, keyword):
709
if isinstance(value, dbus.Boolean):
710
return "Yes" if value else "No"
711
if keyword in ("Timeout", "Interval", "ApprovalDelay",
712
"ApprovalDuration", "ExtendedTimeout"):
713
return cls.milliseconds_to_string(value)
716
def header_line(self, format_string):
717
return format_string.format(**self.tableheaders)
719
def client_line(self, client, format_string):
720
return format_string.format(
721
**{key: self.string_from_client(client, key)
722
for key in self.keywords})
725
def milliseconds_to_string(ms):
726
td = datetime.timedelta(0, 0, 0, ms)
727
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
728
.format(days="{}T".format(td.days)
730
hours=td.seconds // 3600,
731
minutes=(td.seconds % 3600) // 60,
732
seconds=td.seconds % 60))
735
class Property(Base):
736
"Abstract class for Actions for setting one client property"
738
def run_on_one_client(self, client, properties):
739
"""Set the Client's D-Bus property"""
740
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
741
client.__dbus_object_path__,
742
dbus.PROPERTIES_IFACE, client_dbus_interface,
743
self.propname, self.value_to_set
744
if not isinstance(self.value_to_set,
746
else bool(self.value_to_set))
747
client.Set(client_dbus_interface, self.propname,
749
dbus_interface=dbus.PROPERTIES_IFACE)
753
raise NotImplementedError()
756
class Enable(Property):
758
value_to_set = dbus.Boolean(True)
761
class Disable(Property):
763
value_to_set = dbus.Boolean(False)
766
class BumpTimeout(Property):
767
propname = "LastCheckedOK"
771
class StartChecker(Property):
772
propname = "CheckerRunning"
773
value_to_set = dbus.Boolean(True)
776
class StopChecker(Property):
777
propname = "CheckerRunning"
778
value_to_set = dbus.Boolean(False)
781
class ApproveByDefault(Property):
782
propname = "ApprovedByDefault"
783
value_to_set = dbus.Boolean(True)
786
class DenyByDefault(Property):
787
propname = "ApprovedByDefault"
788
value_to_set = dbus.Boolean(False)
791
class PropertyValue(Property):
792
"Abstract class for Property recieving a value as argument"
793
def __init__(self, value):
794
self.value_to_set = value
797
class SetChecker(PropertyValue):
801
class SetHost(PropertyValue):
805
class SetSecret(PropertyValue):
809
def value_to_set(self):
813
def value_to_set(self, value):
814
"""When setting, read data from supplied file object"""
815
self._vts = value.read()
819
class MillisecondsPropertyValueArgument(PropertyValue):
820
"""Abstract class for PropertyValue taking a value argument as
821
a datetime.timedelta() but should store it as milliseconds."""
824
def value_to_set(self):
828
def value_to_set(self, value):
829
"When setting, convert value from a datetime.timedelta"
830
self._vts = int(round(value.total_seconds() * 1000))
833
class SetTimeout(MillisecondsPropertyValueArgument):
837
class SetExtendedTimeout(MillisecondsPropertyValueArgument):
838
propname = "ExtendedTimeout"
841
class SetInterval(MillisecondsPropertyValueArgument):
842
propname = "Interval"
845
class SetApprovalDelay(MillisecondsPropertyValueArgument):
846
propname = "ApprovalDelay"
849
class SetApprovalDuration(MillisecondsPropertyValueArgument):
850
propname = "ApprovalDuration"
738
dbus_filter = NullFilter()
740
dbus_logger.addFilter(dbus_filter)
741
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,
742
server_path, dbus.OBJECT_MANAGER_IFACE)
743
mandos_clients = {path: ifs_and_props[client_interface]
744
for path, ifs_and_props in
745
mandos_serv_object_manager
746
.GetManagedObjects().items()
747
if client_interface in ifs_and_props}
748
except dbus.exceptions.DBusException as e:
749
log.critical("Failed to access Mandos server through D-Bus:"
753
# restore dbus logger
754
dbus_logger.removeFilter(dbus_filter)
756
# Compile dict of (clients: properties) to process
760
clients = {bus.get_object(busname, path): properties
761
for path, properties in mandos_clients.items()}
763
for name in clientnames:
764
for path, client in mandos_clients.items():
765
if client["Name"] == name:
766
client_objc = bus.get_object(busname, path)
767
clients[client_objc] = client
770
log.critical("Client not found on server: %r", name)
773
# Run all commands on clients
774
commands = commands_from_options(options)
775
for command in commands:
776
command.run(mandos_serv, clients)
854
class TestCaseWithAssertLogs(unittest.TestCase):
855
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
857
if not hasattr(unittest.TestCase, "assertLogs"):
858
@contextlib.contextmanager
859
def assertLogs(self, logger, level=logging.INFO):
860
capturing_handler = self.CapturingLevelHandler(level)
861
old_level = logger.level
862
old_propagate = logger.propagate
863
logger.addHandler(capturing_handler)
864
logger.setLevel(level)
865
logger.propagate = False
867
yield capturing_handler.watcher
869
logger.propagate = old_propagate
870
logger.removeHandler(capturing_handler)
871
logger.setLevel(old_level)
872
self.assertGreater(len(capturing_handler.watcher.records),
875
class CapturingLevelHandler(logging.Handler):
876
def __init__(self, level, *args, **kwargs):
877
logging.Handler.__init__(self, *args, **kwargs)
878
self.watcher = self.LoggingWatcher([], [])
879
def emit(self, record):
880
self.watcher.records.append(record)
881
self.watcher.output.append(self.format(record))
883
LoggingWatcher = collections.namedtuple("LoggingWatcher",
888
class Test_string_to_delta(TestCaseWithAssertLogs):
889
# Just test basic RFC 3339 functionality here, the doc string for
890
# rfc3339_duration_to_delta() already has more comprehensive
891
# tests, which is run by doctest.
893
def test_rfc3339_zero_seconds(self):
779
class Test_milliseconds_to_string(unittest.TestCase):
781
self.assertEqual(milliseconds_to_string(93785000),
783
def test_no_days(self):
784
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
785
def test_all_zero(self):
786
self.assertEqual(milliseconds_to_string(0), "00:00:00")
787
def test_no_fractional_seconds(self):
788
self.assertEqual(milliseconds_to_string(400), "00:00:00")
789
self.assertEqual(milliseconds_to_string(900), "00:00:00")
790
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
792
class Test_string_to_delta(unittest.TestCase):
793
def test_handles_basic_rfc3339(self):
894
794
self.assertEqual(string_to_delta("PT0S"),
895
795
datetime.timedelta())
897
def test_rfc3339_zero_days(self):
898
796
self.assertEqual(string_to_delta("P0D"),
899
797
datetime.timedelta())
901
def test_rfc3339_one_second(self):
902
798
self.assertEqual(string_to_delta("PT1S"),
903
799
datetime.timedelta(0, 1))
905
def test_rfc3339_two_hours(self):
906
800
self.assertEqual(string_to_delta("PT2H"),
907
801
datetime.timedelta(0, 7200))
909
802
def test_falls_back_to_pre_1_6_1_with_warning(self):
910
with self.assertLogs(log, logging.WARNING):
911
value = string_to_delta("2h")
803
# assertLogs only exists in Python 3.4
804
if hasattr(self, "assertLogs"):
805
with self.assertLogs(log, logging.WARNING):
806
value = string_to_delta("2h")
808
class WarningFilter(logging.Filter):
809
"""Don't show, but record the presence of, warnings"""
810
def filter(self, record):
811
is_warning = record.levelno >= logging.WARNING
812
self.found = is_warning or getattr(self, "found",
814
return not is_warning
815
warning_filter = WarningFilter()
816
log.addFilter(warning_filter)
818
value = string_to_delta("2h")
820
log.removeFilter(warning_filter)
821
self.assertTrue(getattr(warning_filter, "found", False))
912
822
self.assertEqual(value, datetime.timedelta(0, 7200))
915
class Test_check_option_syntax(unittest.TestCase):
917
self.parser = argparse.ArgumentParser()
918
add_command_line_options(self.parser)
920
def test_actions_requires_client_or_all(self):
921
for action, value in self.actions.items():
922
options = self.parser.parse_args()
923
setattr(options, action, value)
924
with self.assertParseError():
925
self.check_option_syntax(options)
927
# This mostly corresponds to the definition from has_actions() in
928
# check_option_syntax()
930
# The actual values set here are not that important, but we do
931
# at least stick to the correct types, even though they are
935
"bump_timeout": True,
936
"start_checker": True,
937
"stop_checker": True,
941
"timeout": datetime.timedelta(),
942
"extended_timeout": datetime.timedelta(),
943
"interval": datetime.timedelta(),
944
"approved_by_default": True,
945
"approval_delay": datetime.timedelta(),
946
"approval_duration": datetime.timedelta(),
948
"secret": io.BytesIO(b"x"),
953
@contextlib.contextmanager
954
def assertParseError(self):
955
with self.assertRaises(SystemExit) as e:
956
with self.redirect_stderr_to_devnull():
958
# Exit code from argparse is guaranteed to be "2". Reference:
959
# https://docs.python.org/3/library
960
# /argparse.html#exiting-methods
961
self.assertEqual(e.exception.code, 2)
964
@contextlib.contextmanager
965
def redirect_stderr_to_devnull():
966
null = os.open(os.path.devnull, os.O_RDWR)
967
stderrcopy = os.dup(sys.stderr.fileno())
968
os.dup2(null, sys.stderr.fileno())
974
os.dup2(stderrcopy, sys.stderr.fileno())
977
def check_option_syntax(self, options):
978
check_option_syntax(self.parser, options)
980
def test_actions_all_conflicts_with_verbose(self):
981
for action, value in self.actions.items():
982
options = self.parser.parse_args()
983
setattr(options, action, value)
985
options.verbose = True
986
with self.assertParseError():
987
self.check_option_syntax(options)
989
def test_actions_with_client_conflicts_with_verbose(self):
990
for action, value in self.actions.items():
991
options = self.parser.parse_args()
992
setattr(options, action, value)
993
options.verbose = True
994
options.client = ["foo"]
995
with self.assertParseError():
996
self.check_option_syntax(options)
998
def test_dump_json_conflicts_with_verbose(self):
999
options = self.parser.parse_args()
1000
options.dump_json = True
1001
options.verbose = True
1002
with self.assertParseError():
1003
self.check_option_syntax(options)
1005
def test_dump_json_conflicts_with_action(self):
1006
for action, value in self.actions.items():
1007
options = self.parser.parse_args()
1008
setattr(options, action, value)
1009
options.dump_json = True
1010
with self.assertParseError():
1011
self.check_option_syntax(options)
1013
def test_all_can_not_be_alone(self):
1014
options = self.parser.parse_args()
1016
with self.assertParseError():
1017
self.check_option_syntax(options)
1019
def test_all_is_ok_with_any_action(self):
1020
for action, value in self.actions.items():
1021
options = self.parser.parse_args()
1022
setattr(options, action, value)
1024
self.check_option_syntax(options)
1026
def test_any_action_is_ok_with_one_client(self):
1027
for action, value in self.actions.items():
1028
options = self.parser.parse_args()
1029
setattr(options, action, value)
1030
options.client = ["foo"]
1031
self.check_option_syntax(options)
1033
def test_one_client_with_all_actions_except_is_enabled(self):
1034
options = self.parser.parse_args()
1035
for action, value in self.actions.items():
1036
if action == "is_enabled":
1038
setattr(options, action, value)
1039
options.client = ["foo"]
1040
self.check_option_syntax(options)
1042
def test_two_clients_with_all_actions_except_is_enabled(self):
1043
options = self.parser.parse_args()
1044
for action, value in self.actions.items():
1045
if action == "is_enabled":
1047
setattr(options, action, value)
1048
options.client = ["foo", "barbar"]
1049
self.check_option_syntax(options)
1051
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1052
for action, value in self.actions.items():
1053
if action == "is_enabled":
1055
options = self.parser.parse_args()
1056
setattr(options, action, value)
1057
options.client = ["foo", "barbar"]
1058
self.check_option_syntax(options)
1060
def test_is_enabled_fails_without_client(self):
1061
options = self.parser.parse_args()
1062
options.is_enabled = True
1063
with self.assertParseError():
1064
self.check_option_syntax(options)
1066
def test_is_enabled_fails_with_two_clients(self):
1067
options = self.parser.parse_args()
1068
options.is_enabled = True
1069
options.client = ["foo", "barbar"]
1070
with self.assertParseError():
1071
self.check_option_syntax(options)
1073
def test_remove_can_only_be_combined_with_action_deny(self):
1074
for action, value in self.actions.items():
1075
if action in {"remove", "deny"}:
1077
options = self.parser.parse_args()
1078
setattr(options, action, value)
1080
options.remove = True
1081
with self.assertParseError():
1082
self.check_option_syntax(options)
1085
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1086
def test_calls_and_returns_get_object_on_bus(self):
1087
class MockBus(object):
1089
def get_object(mockbus_self, busname, dbus_path):
1090
# Note that "self" is still the testcase instance,
1091
# this MockBus instance is in "mockbus_self".
1092
self.assertEqual(busname, dbus_busname)
1093
self.assertEqual(dbus_path, server_dbus_path)
1094
mockbus_self.called = True
1097
mockbus = get_mandos_dbus_object(bus=MockBus())
1098
self.assertIsInstance(mockbus, MockBus)
1099
self.assertTrue(mockbus.called)
1101
def test_logs_and_exits_on_dbus_error(self):
1102
class MockBusFailing(object):
1103
def get_object(self, busname, dbus_path):
1104
raise dbus.exceptions.DBusException("Test")
1106
with self.assertLogs(log, logging.CRITICAL):
1107
with self.assertRaises(SystemExit) as e:
1108
bus = get_mandos_dbus_object(bus=MockBusFailing())
1110
if isinstance(e.exception.code, int):
1111
self.assertNotEqual(e.exception.code, 0)
1113
self.assertIsNotNone(e.exception.code)
1116
class Test_get_managed_objects(TestCaseWithAssertLogs):
1117
def test_calls_and_returns_GetManagedObjects(self):
1118
managed_objects = {"/clients/foo": { "Name": "foo"}}
1119
class MockObjectManager(object):
1120
def GetManagedObjects(self):
1121
return managed_objects
1122
retval = get_managed_objects(MockObjectManager())
1123
self.assertDictEqual(managed_objects, retval)
1125
def test_logs_and_exits_on_dbus_error(self):
1126
dbus_logger = logging.getLogger("dbus.proxies")
1128
class MockObjectManagerFailing(object):
1129
def GetManagedObjects(self):
1130
dbus_logger.error("Test")
1131
raise dbus.exceptions.DBusException("Test")
1133
class CountingHandler(logging.Handler):
1135
def emit(self, record):
1138
counting_handler = CountingHandler()
1140
dbus_logger.addHandler(counting_handler)
1143
with self.assertLogs(log, logging.CRITICAL) as watcher:
1144
with self.assertRaises(SystemExit) as e:
1145
get_managed_objects(MockObjectManagerFailing())
1147
dbus_logger.removeFilter(counting_handler)
1149
# Make sure the dbus logger was suppressed
1150
self.assertEqual(counting_handler.count, 0)
1152
# Test that the dbus_logger still works
1153
with self.assertLogs(dbus_logger, logging.ERROR):
1154
dbus_logger.error("Test")
1156
if isinstance(e.exception.code, int):
1157
self.assertNotEqual(e.exception.code, 0)
1159
self.assertIsNotNone(e.exception.code)
1162
class Test_commands_from_options(unittest.TestCase):
1164
self.parser = argparse.ArgumentParser()
1165
add_command_line_options(self.parser)
1167
def test_is_enabled(self):
1168
self.assert_command_from_args(["--is-enabled", "foo"],
1171
def assert_command_from_args(self, args, command_cls,
1173
"""Assert that parsing ARGS should result in an instance of
1174
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1175
options = self.parser.parse_args(args)
1176
check_option_syntax(self.parser, options)
1177
commands = commands_from_options(options)
1178
self.assertEqual(len(commands), 1)
1179
command = commands[0]
1180
self.assertIsInstance(command, command_cls)
1181
for key, value in cmd_attrs.items():
1182
self.assertEqual(getattr(command, key), value)
1184
def test_is_enabled_short(self):
1185
self.assert_command_from_args(["-V", "foo"],
1188
def test_approve(self):
1189
self.assert_command_from_args(["--approve", "foo"],
1192
def test_approve_short(self):
1193
self.assert_command_from_args(["-A", "foo"], command.Approve)
1195
def test_deny(self):
1196
self.assert_command_from_args(["--deny", "foo"], command.Deny)
1198
def test_deny_short(self):
1199
self.assert_command_from_args(["-D", "foo"], command.Deny)
1201
def test_remove(self):
1202
self.assert_command_from_args(["--remove", "foo"],
1205
def test_deny_before_remove(self):
1206
options = self.parser.parse_args(["--deny", "--remove",
1208
check_option_syntax(self.parser, options)
1209
commands = commands_from_options(options)
1210
self.assertEqual(len(commands), 2)
1211
self.assertIsInstance(commands[0], command.Deny)
1212
self.assertIsInstance(commands[1], command.Remove)
1214
def test_deny_before_remove_reversed(self):
1215
options = self.parser.parse_args(["--remove", "--deny",
1217
check_option_syntax(self.parser, options)
1218
commands = commands_from_options(options)
1219
self.assertEqual(len(commands), 2)
1220
self.assertIsInstance(commands[0], command.Deny)
1221
self.assertIsInstance(commands[1], command.Remove)
1223
def test_remove_short(self):
1224
self.assert_command_from_args(["-r", "foo"], command.Remove)
1226
def test_dump_json(self):
1227
self.assert_command_from_args(["--dump-json"],
1230
def test_enable(self):
1231
self.assert_command_from_args(["--enable", "foo"],
1234
def test_enable_short(self):
1235
self.assert_command_from_args(["-e", "foo"], command.Enable)
1237
def test_disable(self):
1238
self.assert_command_from_args(["--disable", "foo"],
1241
def test_disable_short(self):
1242
self.assert_command_from_args(["-d", "foo"], command.Disable)
1244
def test_bump_timeout(self):
1245
self.assert_command_from_args(["--bump-timeout", "foo"],
1246
command.BumpTimeout)
1248
def test_bump_timeout_short(self):
1249
self.assert_command_from_args(["-b", "foo"],
1250
command.BumpTimeout)
1252
def test_start_checker(self):
1253
self.assert_command_from_args(["--start-checker", "foo"],
1254
command.StartChecker)
1256
def test_stop_checker(self):
1257
self.assert_command_from_args(["--stop-checker", "foo"],
1258
command.StopChecker)
1260
def test_approve_by_default(self):
1261
self.assert_command_from_args(["--approve-by-default", "foo"],
1262
command.ApproveByDefault)
1264
def test_deny_by_default(self):
1265
self.assert_command_from_args(["--deny-by-default", "foo"],
1266
command.DenyByDefault)
1268
def test_checker(self):
1269
self.assert_command_from_args(["--checker", ":", "foo"],
1273
def test_checker_empty(self):
1274
self.assert_command_from_args(["--checker", "", "foo"],
1278
def test_checker_short(self):
1279
self.assert_command_from_args(["-c", ":", "foo"],
1283
def test_host(self):
1284
self.assert_command_from_args(["--host", "foo.example.org",
1285
"foo"], command.SetHost,
1286
value_to_set="foo.example.org")
1288
def test_host_short(self):
1289
self.assert_command_from_args(["-H", "foo.example.org",
1290
"foo"], command.SetHost,
1291
value_to_set="foo.example.org")
1293
def test_secret_devnull(self):
1294
self.assert_command_from_args(["--secret", os.path.devnull,
1295
"foo"], command.SetSecret,
1298
def test_secret_tempfile(self):
1299
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1300
value = b"secret\0xyzzy\nbar"
1303
self.assert_command_from_args(["--secret", f.name,
1304
"foo"], command.SetSecret,
1307
def test_secret_devnull_short(self):
1308
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1312
def test_secret_tempfile_short(self):
1313
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1314
value = b"secret\0xyzzy\nbar"
1317
self.assert_command_from_args(["-s", f.name, "foo"],
1321
def test_timeout(self):
1322
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1324
value_to_set=300000)
1326
def test_timeout_short(self):
1327
self.assert_command_from_args(["-t", "PT5M", "foo"],
1329
value_to_set=300000)
1331
def test_extended_timeout(self):
1332
self.assert_command_from_args(["--extended-timeout", "PT15M",
1334
command.SetExtendedTimeout,
1335
value_to_set=900000)
1337
def test_interval(self):
1338
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1339
command.SetInterval,
1340
value_to_set=120000)
1342
def test_interval_short(self):
1343
self.assert_command_from_args(["-i", "PT2M", "foo"],
1344
command.SetInterval,
1345
value_to_set=120000)
1347
def test_approval_delay(self):
1348
self.assert_command_from_args(["--approval-delay", "PT30S",
1350
command.SetApprovalDelay,
1353
def test_approval_duration(self):
1354
self.assert_command_from_args(["--approval-duration", "PT1S",
1356
command.SetApprovalDuration,
1359
def test_print_table(self):
1360
self.assert_command_from_args([], command.PrintTable,
1363
def test_print_table_verbose(self):
1364
self.assert_command_from_args(["--verbose"],
1368
def test_print_table_verbose_short(self):
1369
self.assert_command_from_args(["-v"], command.PrintTable,
1373
class TestCommand(unittest.TestCase):
825
class TestCmd(unittest.TestCase):
1374
826
"""Abstract class for tests of command classes"""
1376
827
def setUp(self):
1378
829
class MockClient(object):
1379
830
def __init__(self, name, **attributes):
1380
self.__dbus_object_path__ = "/clients/{}".format(name)
831
self.__dbus_object_path__ = "objpath_{}".format(name)
1381
832
self.attributes = attributes
1382
833
self.attributes["Name"] = name
1384
def Set(self, interface, propname, value, dbus_interface):
1385
testcase.assertEqual(interface, client_dbus_interface)
1386
testcase.assertEqual(dbus_interface,
1387
dbus.PROPERTIES_IFACE)
1388
self.attributes[propname] = value
835
def Set(self, interface, property, value, dbus_interface):
836
testcase.assertEqual(interface, client_interface)
837
testcase.assertEqual(dbus_interface,
838
dbus.PROPERTIES_IFACE)
839
self.attributes[property] = value
840
def Get(self, interface, property, dbus_interface):
841
testcase.assertEqual(interface, client_interface)
842
testcase.assertEqual(dbus_interface,
843
dbus.PROPERTIES_IFACE)
844
return self.attributes[property]
1389
845
def Approve(self, approve, dbus_interface):
1390
testcase.assertEqual(dbus_interface,
1391
client_dbus_interface)
846
testcase.assertEqual(dbus_interface, client_interface)
1392
847
self.calls.append(("Approve", (approve,
1393
848
dbus_interface)))
1394
849
self.client = MockClient(
1434
889
ApprovedByDefault=dbus.Boolean(False),
1435
890
LastApprovalRequest="2019-01-03T00:00:00",
1436
891
ApprovalDelay=30000,
1437
ApprovalDuration=93785000,
892
ApprovalDuration=1000,
1439
894
ExtendedTimeout=900000,
1440
895
Expires="2019-02-05T00:00:00",
1441
896
LastCheckerStatus=-2)
1442
897
self.clients = collections.OrderedDict(
1444
("/clients/foo", self.client.attributes),
1445
("/clients/barbar", self.other_client.attributes),
899
(self.client, self.client.attributes),
900
(self.other_client, self.other_client.attributes),
1447
self.one_client = {"/clients/foo": self.client.attributes}
1453
def get_object(client_bus_name, path):
1454
self.assertEqual(client_bus_name, dbus_busname)
1456
# Note: "self" here is the TestCmd instance, not
1457
# the Bus instance, since this is a static method!
1458
"/clients/foo": self.client,
1459
"/clients/barbar": self.other_client,
1464
class TestBaseCommands(TestCommand):
1466
def test_IsEnabled_exits_successfully(self):
902
self.one_client = {self.client: self.client.attributes}
904
class TestPrintTableCmd(TestCmd):
905
def test_normal(self):
906
output = PrintTableCmd().output(self.clients)
907
expected_output = """
908
Name Enabled Timeout Last Successful Check
909
foo Yes 00:05:00 2019-02-03T00:00:00
910
barbar Yes 00:05:00 2019-02-04T00:00:00
912
self.assertEqual(output, expected_output)
913
def test_verbose(self):
914
output = PrintTableCmd(verbose=True).output(self.clients)
915
expected_output = """
916
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
917
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
918
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
920
self.assertEqual(output, expected_output)
921
def test_one_client(self):
922
output = PrintTableCmd().output(self.one_client)
923
expected_output = """
924
Name Enabled Timeout Last Successful Check
925
foo Yes 00:05:00 2019-02-03T00:00:00
927
self.assertEqual(output, expected_output)
929
class TestDumpJSONCmd(TestCmd):
931
self.expected_json = {
934
"KeyID": ("92ed150794387c03ce684574b1139a65"
935
"94a34f895daaaf09fd8ea90a27cddb12"),
936
"Host": "foo.example.org",
939
"LastCheckedOK": "2019-02-03T00:00:00",
940
"Created": "2019-01-02T00:00:00",
942
"Fingerprint": ("778827225BA7DE539C5A"
943
"7CFA59CFF7CDBD9A5920"),
944
"CheckerRunning": False,
945
"LastEnabled": "2019-01-03T00:00:00",
946
"ApprovalPending": False,
947
"ApprovedByDefault": True,
948
"LastApprovalRequest": "",
950
"ApprovalDuration": 1000,
951
"Checker": "fping -q -- %(host)s",
952
"ExtendedTimeout": 900000,
953
"Expires": "2019-02-04T00:00:00",
954
"LastCheckerStatus": 0,
958
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
959
"6ab612cff5ad227247e46c2b020f441c"),
963
"LastCheckedOK": "2019-02-04T00:00:00",
964
"Created": "2019-01-03T00:00:00",
966
"Fingerprint": ("3E393AEAEFB84C7E89E2"
967
"F547B3A107558FCA3A27"),
968
"CheckerRunning": True,
969
"LastEnabled": "2019-01-04T00:00:00",
970
"ApprovalPending": False,
971
"ApprovedByDefault": False,
972
"LastApprovalRequest": "2019-01-03T00:00:00",
973
"ApprovalDelay": 30000,
974
"ApprovalDuration": 1000,
976
"ExtendedTimeout": 900000,
977
"Expires": "2019-02-05T00:00:00",
978
"LastCheckerStatus": -2,
981
return super(TestDumpJSONCmd, self).setUp()
982
def test_normal(self):
983
json_data = json.loads(DumpJSONCmd().output(self.clients))
984
self.assertDictEqual(json_data, self.expected_json)
985
def test_one_client(self):
986
clients = self.one_client
987
json_data = json.loads(DumpJSONCmd().output(clients))
988
expected_json = {"foo": self.expected_json["foo"]}
989
self.assertDictEqual(json_data, expected_json)
991
class TestIsEnabledCmd(TestCmd):
992
def test_is_enabled(self):
993
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
994
for client, properties in self.clients.items()))
995
def test_is_enabled_run_exits_successfully(self):
1467
996
with self.assertRaises(SystemExit) as e:
1468
command.IsEnabled().run(self.one_client)
997
IsEnabledCmd().run(None, self.one_client)
1469
998
if e.exception.code is not None:
1470
999
self.assertEqual(e.exception.code, 0)
1472
1001
self.assertIsNone(e.exception.code)
1474
def test_IsEnabled_exits_with_failure(self):
1002
def test_is_enabled_run_exits_with_failure(self):
1475
1003
self.client.attributes["Enabled"] = dbus.Boolean(False)
1476
1004
with self.assertRaises(SystemExit) as e:
1477
command.IsEnabled().run(self.one_client)
1005
IsEnabledCmd().run(None, self.one_client)
1478
1006
if isinstance(e.exception.code, int):
1479
1007
self.assertNotEqual(e.exception.code, 0)
1481
1009
self.assertIsNotNone(e.exception.code)
1483
def test_Approve(self):
1484
command.Approve().run(self.clients, self.bus)
1485
for clientpath in self.clients:
1486
client = self.bus.get_object(dbus_busname, clientpath)
1487
self.assertIn(("Approve", (True, client_dbus_interface)),
1490
def test_Deny(self):
1491
command.Deny().run(self.clients, self.bus)
1492
for clientpath in self.clients:
1493
client = self.bus.get_object(dbus_busname, clientpath)
1494
self.assertIn(("Approve", (False, client_dbus_interface)),
1497
def test_Remove(self):
1011
class TestRemoveCmd(TestCmd):
1012
def test_remove(self):
1498
1013
class MockMandos(object):
1499
1014
def __init__(self):
1500
1015
self.calls = []
1501
1016
def RemoveClient(self, dbus_path):
1502
1017
self.calls.append(("RemoveClient", (dbus_path,)))
1503
1018
mandos = MockMandos()
1504
command.Remove().run(self.clients, self.bus, mandos)
1505
for clientpath in self.clients:
1506
self.assertIn(("RemoveClient", (clientpath,)),
1019
super(TestRemoveCmd, self).setUp()
1020
RemoveCmd().run(mandos, self.clients)
1021
self.assertEqual(len(mandos.calls), 2)
1022
for client in self.clients:
1023
self.assertIn(("RemoveClient",
1024
(client.__dbus_object_path__,)),
1512
"KeyID": ("92ed150794387c03ce684574b1139a65"
1513
"94a34f895daaaf09fd8ea90a27cddb12"),
1514
"Host": "foo.example.org",
1517
"LastCheckedOK": "2019-02-03T00:00:00",
1518
"Created": "2019-01-02T00:00:00",
1520
"Fingerprint": ("778827225BA7DE539C5A"
1521
"7CFA59CFF7CDBD9A5920"),
1522
"CheckerRunning": False,
1523
"LastEnabled": "2019-01-03T00:00:00",
1524
"ApprovalPending": False,
1525
"ApprovedByDefault": True,
1526
"LastApprovalRequest": "",
1528
"ApprovalDuration": 1000,
1529
"Checker": "fping -q -- %(host)s",
1530
"ExtendedTimeout": 900000,
1531
"Expires": "2019-02-04T00:00:00",
1532
"LastCheckerStatus": 0,
1536
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1537
"6ab612cff5ad227247e46c2b020f441c"),
1538
"Host": "192.0.2.3",
1541
"LastCheckedOK": "2019-02-04T00:00:00",
1542
"Created": "2019-01-03T00:00:00",
1544
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1545
"F547B3A107558FCA3A27"),
1546
"CheckerRunning": True,
1547
"LastEnabled": "2019-01-04T00:00:00",
1548
"ApprovalPending": False,
1549
"ApprovedByDefault": False,
1550
"LastApprovalRequest": "2019-01-03T00:00:00",
1551
"ApprovalDelay": 30000,
1552
"ApprovalDuration": 93785000,
1554
"ExtendedTimeout": 900000,
1555
"Expires": "2019-02-05T00:00:00",
1556
"LastCheckerStatus": -2,
1560
def test_DumpJSON_normal(self):
1561
output = command.DumpJSON().output(self.clients.values())
1562
json_data = json.loads(output)
1563
self.assertDictEqual(json_data, self.expected_json)
1565
def test_DumpJSON_one_client(self):
1566
output = command.DumpJSON().output(self.one_client.values())
1567
json_data = json.loads(output)
1568
expected_json = {"foo": self.expected_json["foo"]}
1569
self.assertDictEqual(json_data, expected_json)
1571
def test_PrintTable_normal(self):
1572
output = command.PrintTable().output(self.clients.values())
1573
expected_output = "\n".join((
1574
"Name Enabled Timeout Last Successful Check",
1575
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1576
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1578
self.assertEqual(output, expected_output)
1580
def test_PrintTable_verbose(self):
1581
output = command.PrintTable(verbose=True).output(
1582
self.clients.values())
1597
"Last Successful Check ",
1598
"2019-02-03T00:00:00 ",
1599
"2019-02-04T00:00:00 ",
1602
"2019-01-02T00:00:00 ",
1603
"2019-01-03T00:00:00 ",
1615
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1617
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1621
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1622
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1624
"Check Is Running ",
1629
"2019-01-03T00:00:00 ",
1630
"2019-01-04T00:00:00 ",
1632
"Approval Is Pending ",
1636
"Approved By Default ",
1640
"Last Approval Request ",
1642
"2019-01-03T00:00:00 ",
1648
"Approval Duration ",
1653
"fping -q -- %(host)s ",
1656
"Extended Timeout ",
1661
"2019-02-04T00:00:00 ",
1662
"2019-02-05T00:00:00 ",
1664
"Last Checker Status",
1669
num_lines = max(len(rows) for rows in columns)
1670
expected_output = "\n".join("".join(rows[line]
1671
for rows in columns)
1672
for line in range(num_lines))
1673
self.assertEqual(output, expected_output)
1675
def test_PrintTable_one_client(self):
1676
output = command.PrintTable().output(self.one_client.values())
1677
expected_output = "\n".join((
1678
"Name Enabled Timeout Last Successful Check",
1679
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1681
self.assertEqual(output, expected_output)
1684
class TestPropertyCmd(TestCommand):
1685
"""Abstract class for tests of command.Property classes"""
1027
class TestApproveCmd(TestCmd):
1028
def test_approve(self):
1029
ApproveCmd().run(None, self.clients)
1030
for client in self.clients:
1031
self.assertIn(("Approve", (True, client_interface)),
1034
class TestDenyCmd(TestCmd):
1035
def test_deny(self):
1036
DenyCmd().run(None, self.clients)
1037
for client in self.clients:
1038
self.assertIn(("Approve", (False, client_interface)),
1041
class TestEnableCmd(TestCmd):
1042
def test_enable(self):
1043
for client in self.clients:
1044
client.attributes["Enabled"] = False
1046
EnableCmd().run(None, self.clients)
1048
for client in self.clients:
1049
self.assertTrue(client.attributes["Enabled"])
1051
class TestDisableCmd(TestCmd):
1052
def test_disable(self):
1053
DisableCmd().run(None, self.clients)
1055
for client in self.clients:
1056
self.assertFalse(client.attributes["Enabled"])
1058
class Unique(object):
1059
"""Class for objects which exist only to be unique objects, since
1060
unittest.mock.sentinel only exists in Python 3.3"""
1062
class TestPropertyCmd(TestCmd):
1063
"""Abstract class for tests of PropertyCmd classes"""
1686
1064
def runTest(self):
1687
1065
if not hasattr(self, "command"):
1690
1068
self.values_to_set)
1691
1069
for value_to_set, value_to_get in zip(self.values_to_set,
1692
1070
values_to_get):
1693
for clientpath in self.clients:
1694
client = self.bus.get_object(dbus_busname, clientpath)
1695
old_value = client.attributes[self.propname]
1696
client.attributes[self.propname] = self.Unique()
1071
for client in self.clients:
1072
old_value = client.attributes[self.property]
1073
self.assertNotIsInstance(old_value, Unique)
1074
client.attributes[self.property] = Unique()
1697
1075
self.run_command(value_to_set, self.clients)
1698
for clientpath in self.clients:
1699
client = self.bus.get_object(dbus_busname, clientpath)
1700
value = client.attributes[self.propname]
1701
self.assertNotIsInstance(value, self.Unique)
1076
for client in self.clients:
1077
value = client.attributes[self.property]
1078
self.assertNotIsInstance(value, Unique)
1702
1079
self.assertEqual(value, value_to_get)
1704
class Unique(object):
1705
"""Class for objects which exist only to be unique objects,
1706
since unittest.mock.sentinel only exists in Python 3.3"""
1708
1080
def run_command(self, value, clients):
1709
self.command().run(clients, self.bus)
1712
class TestEnableCmd(TestPropertyCmd):
1713
command = command.Enable
1714
propname = "Enabled"
1715
values_to_set = [dbus.Boolean(True)]
1718
class TestDisableCmd(TestPropertyCmd):
1719
command = command.Disable
1720
propname = "Enabled"
1721
values_to_set = [dbus.Boolean(False)]
1081
self.command().run(None, clients)
1724
1083
class TestBumpTimeoutCmd(TestPropertyCmd):
1725
command = command.BumpTimeout
1726
propname = "LastCheckedOK"
1084
command = BumpTimeoutCmd
1085
property = "LastCheckedOK"
1727
1086
values_to_set = [""]
1730
1088
class TestStartCheckerCmd(TestPropertyCmd):
1731
command = command.StartChecker
1732
propname = "CheckerRunning"
1089
command = StartCheckerCmd
1090
property = "CheckerRunning"
1733
1091
values_to_set = [dbus.Boolean(True)]
1736
1093
class TestStopCheckerCmd(TestPropertyCmd):
1737
command = command.StopChecker
1738
propname = "CheckerRunning"
1094
command = StopCheckerCmd
1095
property = "CheckerRunning"
1739
1096
values_to_set = [dbus.Boolean(False)]
1742
1098
class TestApproveByDefaultCmd(TestPropertyCmd):
1743
command = command.ApproveByDefault
1744
propname = "ApprovedByDefault"
1099
command = ApproveByDefaultCmd
1100
property = "ApprovedByDefault"
1745
1101
values_to_set = [dbus.Boolean(True)]
1748
1103
class TestDenyByDefaultCmd(TestPropertyCmd):
1749
command = command.DenyByDefault
1750
propname = "ApprovedByDefault"
1104
command = DenyByDefaultCmd
1105
property = "ApprovedByDefault"
1751
1106
values_to_set = [dbus.Boolean(False)]
1754
class TestPropertyValueCmd(TestPropertyCmd):
1755
"""Abstract class for tests of PropertyValueCmd classes"""
1108
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1109
"""Abstract class for tests of PropertyCmd classes using the
1110
ValueArgumentMixIn"""
1757
1111
def runTest(self):
1758
if type(self) is TestPropertyValueCmd:
1112
if type(self) is TestValueArgumentPropertyCmd:
1760
return super(TestPropertyValueCmd, self).runTest()
1114
return super(TestValueArgumentPropertyCmd, self).runTest()
1762
1115
def run_command(self, value, clients):
1763
self.command(value).run(clients, self.bus)
1766
class TestSetCheckerCmd(TestPropertyValueCmd):
1767
command = command.SetChecker
1768
propname = "Checker"
1116
self.command(value).run(None, clients)
1118
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1119
command = SetCheckerCmd
1120
property = "Checker"
1769
1121
values_to_set = ["", ":", "fping -q -- %s"]
1772
class TestSetHostCmd(TestPropertyValueCmd):
1773
command = command.SetHost
1123
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1124
command = SetHostCmd
1775
1126
values_to_set = ["192.0.2.3", "foo.example.org"]
1778
class TestSetSecretCmd(TestPropertyValueCmd):
1779
command = command.SetSecret
1128
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1129
command = SetSecretCmd
1781
1131
values_to_set = [io.BytesIO(b""),
1782
1132
io.BytesIO(b"secret\0xyzzy\nbar")]
1783
1133
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1786
class TestSetTimeoutCmd(TestPropertyValueCmd):
1787
command = command.SetTimeout
1788
propname = "Timeout"
1789
values_to_set = [datetime.timedelta(),
1790
datetime.timedelta(minutes=5),
1791
datetime.timedelta(seconds=1),
1792
datetime.timedelta(weeks=1),
1793
datetime.timedelta(weeks=52)]
1794
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1797
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1798
command = command.SetExtendedTimeout
1799
propname = "ExtendedTimeout"
1800
values_to_set = [datetime.timedelta(),
1801
datetime.timedelta(minutes=5),
1802
datetime.timedelta(seconds=1),
1803
datetime.timedelta(weeks=1),
1804
datetime.timedelta(weeks=52)]
1805
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1808
class TestSetIntervalCmd(TestPropertyValueCmd):
1809
command = command.SetInterval
1810
propname = "Interval"
1811
values_to_set = [datetime.timedelta(),
1812
datetime.timedelta(minutes=5),
1813
datetime.timedelta(seconds=1),
1814
datetime.timedelta(weeks=1),
1815
datetime.timedelta(weeks=52)]
1816
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1819
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1820
command = command.SetApprovalDelay
1821
propname = "ApprovalDelay"
1822
values_to_set = [datetime.timedelta(),
1823
datetime.timedelta(minutes=5),
1824
datetime.timedelta(seconds=1),
1825
datetime.timedelta(weeks=1),
1826
datetime.timedelta(weeks=52)]
1827
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1830
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1831
command = command.SetApprovalDuration
1832
propname = "ApprovalDuration"
1833
values_to_set = [datetime.timedelta(),
1834
datetime.timedelta(minutes=5),
1835
datetime.timedelta(seconds=1),
1836
datetime.timedelta(weeks=1),
1837
datetime.timedelta(weeks=52)]
1838
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1135
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1136
command = SetTimeoutCmd
1137
property = "Timeout"
1138
values_to_set = [datetime.timedelta(),
1139
datetime.timedelta(minutes=5),
1140
datetime.timedelta(seconds=1),
1141
datetime.timedelta(weeks=1),
1142
datetime.timedelta(weeks=52)]
1143
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1145
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1146
command = SetExtendedTimeoutCmd
1147
property = "ExtendedTimeout"
1148
values_to_set = [datetime.timedelta(),
1149
datetime.timedelta(minutes=5),
1150
datetime.timedelta(seconds=1),
1151
datetime.timedelta(weeks=1),
1152
datetime.timedelta(weeks=52)]
1153
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1155
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1156
command = SetIntervalCmd
1157
property = "Interval"
1158
values_to_set = [datetime.timedelta(),
1159
datetime.timedelta(minutes=5),
1160
datetime.timedelta(seconds=1),
1161
datetime.timedelta(weeks=1),
1162
datetime.timedelta(weeks=52)]
1163
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1165
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1166
command = SetApprovalDelayCmd
1167
property = "ApprovalDelay"
1168
values_to_set = [datetime.timedelta(),
1169
datetime.timedelta(minutes=5),
1170
datetime.timedelta(seconds=1),
1171
datetime.timedelta(weeks=1),
1172
datetime.timedelta(weeks=52)]
1173
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1175
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1176
command = SetApprovalDurationCmd
1177
property = "ApprovalDuration"
1178
values_to_set = [datetime.timedelta(),
1179
datetime.timedelta(minutes=5),
1180
datetime.timedelta(seconds=1),
1181
datetime.timedelta(weeks=1),
1182
datetime.timedelta(weeks=52)]
1183
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1185
class Test_command_from_options(unittest.TestCase):
1187
self.parser = argparse.ArgumentParser()
1188
add_command_line_options(self.parser)
1189
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1190
"""Assert that parsing ARGS should result in an instance of
1191
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1192
options = self.parser.parse_args(args)
1193
check_option_syntax(self.parser, options)
1194
commands = commands_from_options(options)
1195
self.assertEqual(len(commands), 1)
1196
command = commands[0]
1197
self.assertIsInstance(command, command_cls)
1198
for key, value in cmd_attrs.items():
1199
self.assertEqual(getattr(command, key), value)
1200
def test_print_table(self):
1201
self.assert_command_from_args([], PrintTableCmd,
1204
def test_print_table_verbose(self):
1205
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1208
def test_print_table_verbose_short(self):
1209
self.assert_command_from_args(["-v"], PrintTableCmd,
1212
def test_enable(self):
1213
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1215
def test_enable_short(self):
1216
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1218
def test_disable(self):
1219
self.assert_command_from_args(["--disable", "foo"],
1222
def test_disable_short(self):
1223
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1225
def test_bump_timeout(self):
1226
self.assert_command_from_args(["--bump-timeout", "foo"],
1229
def test_bump_timeout_short(self):
1230
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1232
def test_start_checker(self):
1233
self.assert_command_from_args(["--start-checker", "foo"],
1236
def test_stop_checker(self):
1237
self.assert_command_from_args(["--stop-checker", "foo"],
1240
def test_remove(self):
1241
self.assert_command_from_args(["--remove", "foo"],
1244
def test_remove_short(self):
1245
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1247
def test_checker(self):
1248
self.assert_command_from_args(["--checker", ":", "foo"],
1249
SetCheckerCmd, value_to_set=":")
1251
def test_checker_empty(self):
1252
self.assert_command_from_args(["--checker", "", "foo"],
1253
SetCheckerCmd, value_to_set="")
1255
def test_checker_short(self):
1256
self.assert_command_from_args(["-c", ":", "foo"],
1257
SetCheckerCmd, value_to_set=":")
1259
def test_timeout(self):
1260
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1262
value_to_set=300000)
1264
def test_timeout_short(self):
1265
self.assert_command_from_args(["-t", "PT5M", "foo"],
1267
value_to_set=300000)
1269
def test_extended_timeout(self):
1270
self.assert_command_from_args(["--extended-timeout", "PT15M",
1272
SetExtendedTimeoutCmd,
1273
value_to_set=900000)
1275
def test_interval(self):
1276
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1278
value_to_set=120000)
1280
def test_interval_short(self):
1281
self.assert_command_from_args(["-i", "PT2M", "foo"],
1283
value_to_set=120000)
1285
def test_approve_by_default(self):
1286
self.assert_command_from_args(["--approve-by-default", "foo"],
1287
ApproveByDefaultCmd)
1289
def test_deny_by_default(self):
1290
self.assert_command_from_args(["--deny-by-default", "foo"],
1293
def test_approval_delay(self):
1294
self.assert_command_from_args(["--approval-delay", "PT30S",
1295
"foo"], SetApprovalDelayCmd,
1298
def test_approval_duration(self):
1299
self.assert_command_from_args(["--approval-duration", "PT1S",
1300
"foo"], SetApprovalDurationCmd,
1303
def test_host(self):
1304
self.assert_command_from_args(["--host", "foo.example.org",
1306
value_to_set="foo.example.org")
1308
def test_host_short(self):
1309
self.assert_command_from_args(["-H", "foo.example.org",
1311
value_to_set="foo.example.org")
1313
def test_secret_devnull(self):
1314
self.assert_command_from_args(["--secret", os.path.devnull,
1315
"foo"], SetSecretCmd,
1318
def test_secret_tempfile(self):
1319
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1320
value = b"secret\0xyzzy\nbar"
1323
self.assert_command_from_args(["--secret", f.name,
1324
"foo"], SetSecretCmd,
1327
def test_secret_devnull_short(self):
1328
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1329
SetSecretCmd, value_to_set=b"")
1331
def test_secret_tempfile_short(self):
1332
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1333
value = b"secret\0xyzzy\nbar"
1336
self.assert_command_from_args(["-s", f.name, "foo"],
1340
def test_approve(self):
1341
self.assert_command_from_args(["--approve", "foo"],
1344
def test_approve_short(self):
1345
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1347
def test_deny(self):
1348
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1350
def test_deny_short(self):
1351
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1353
def test_dump_json(self):
1354
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1356
def test_is_enabled(self):
1357
self.assert_command_from_args(["--is-enabled", "foo"],
1360
def test_is_enabled_short(self):
1361
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1363
def test_deny_before_remove(self):
1364
options = self.parser.parse_args(["--deny", "--remove", "foo"])
1365
check_option_syntax(self.parser, options)
1366
commands = commands_from_options(options)
1367
self.assertEqual(len(commands), 2)
1368
self.assertIsInstance(commands[0], DenyCmd)
1369
self.assertIsInstance(commands[1], RemoveCmd)
1371
def test_deny_before_remove_reversed(self):
1372
options = self.parser.parse_args(["--remove", "--deny", "--all"])
1373
check_option_syntax(self.parser, options)
1374
commands = commands_from_options(options)
1375
self.assertEqual(len(commands), 2)
1376
self.assertIsInstance(commands[0], DenyCmd)
1377
self.assertIsInstance(commands[1], RemoveCmd)
1380
class Test_check_option_syntax(unittest.TestCase):
1381
# This mostly corresponds to the definition from has_actions() in
1382
# check_option_syntax()
1384
# The actual values set here are not that important, but we do
1385
# at least stick to the correct types, even though they are
1389
"bump_timeout": True,
1390
"start_checker": True,
1391
"stop_checker": True,
1395
"timeout": datetime.timedelta(),
1396
"extended_timeout": datetime.timedelta(),
1397
"interval": datetime.timedelta(),
1398
"approved_by_default": True,
1399
"approval_delay": datetime.timedelta(),
1400
"approval_duration": datetime.timedelta(),
1402
"secret": io.BytesIO(b"x"),
1408
self.parser = argparse.ArgumentParser()
1409
add_command_line_options(self.parser)
1411
@contextlib.contextmanager
1412
def assertParseError(self):
1413
with self.assertRaises(SystemExit) as e:
1414
with self.temporarily_suppress_stderr():
1416
# Exit code from argparse is guaranteed to be "2". Reference:
1417
# https://docs.python.org/3/library/argparse.html#exiting-methods
1418
self.assertEqual(e.exception.code, 2)
1421
@contextlib.contextmanager
1422
def temporarily_suppress_stderr():
1423
null = os.open(os.path.devnull, os.O_RDWR)
1424
stderrcopy = os.dup(sys.stderr.fileno())
1425
os.dup2(null, sys.stderr.fileno())
1431
os.dup2(stderrcopy, sys.stderr.fileno())
1432
os.close(stderrcopy)
1434
def check_option_syntax(self, options):
1435
check_option_syntax(self.parser, options)
1437
def test_actions_requires_client_or_all(self):
1438
for action, value in self.actions.items():
1439
options = self.parser.parse_args()
1440
setattr(options, action, value)
1441
with self.assertParseError():
1442
self.check_option_syntax(options)
1444
def test_actions_conflicts_with_verbose(self):
1445
for action, value in self.actions.items():
1446
options = self.parser.parse_args()
1447
setattr(options, action, value)
1448
options.verbose = True
1449
with self.assertParseError():
1450
self.check_option_syntax(options)
1452
def test_dump_json_conflicts_with_verbose(self):
1453
options = self.parser.parse_args()
1454
options.dump_json = True
1455
options.verbose = True
1456
with self.assertParseError():
1457
self.check_option_syntax(options)
1459
def test_dump_json_conflicts_with_action(self):
1460
for action, value in self.actions.items():
1461
options = self.parser.parse_args()
1462
setattr(options, action, value)
1463
options.dump_json = True
1464
with self.assertParseError():
1465
self.check_option_syntax(options)
1467
def test_all_can_not_be_alone(self):
1468
options = self.parser.parse_args()
1470
with self.assertParseError():
1471
self.check_option_syntax(options)
1473
def test_all_is_ok_with_any_action(self):
1474
for action, value in self.actions.items():
1475
options = self.parser.parse_args()
1476
setattr(options, action, value)
1478
self.check_option_syntax(options)
1480
def test_is_enabled_fails_without_client(self):
1481
options = self.parser.parse_args()
1482
options.is_enabled = True
1483
with self.assertParseError():
1484
self.check_option_syntax(options)
1486
def test_is_enabled_works_with_one_client(self):
1487
options = self.parser.parse_args()
1488
options.is_enabled = True
1489
options.client = ["foo"]
1490
self.check_option_syntax(options)
1492
def test_is_enabled_fails_with_two_clients(self):
1493
options = self.parser.parse_args()
1494
options.is_enabled = True
1495
options.client = ["foo", "barbar"]
1496
with self.assertParseError():
1497
self.check_option_syntax(options)
1499
def test_remove_can_only_be_combined_with_action_deny(self):
1500
for action, value in self.actions.items():
1501
if action in {"remove", "deny"}:
1503
options = self.parser.parse_args()
1504
setattr(options, action, value)
1506
options.remove = True
1507
with self.assertParseError():
1508
self.check_option_syntax(options)