462
481
if options.is_enabled:
463
commands.append(IsEnabledCmd())
482
commands.append(command.IsEnabled())
465
484
if options.approve:
466
commands.append(ApproveCmd())
485
commands.append(command.Approve())
469
commands.append(DenyCmd())
488
commands.append(command.Deny())
471
490
if options.remove:
472
commands.append(RemoveCmd())
491
commands.append(command.Remove())
474
493
if options.dump_json:
475
commands.append(DumpJSONCmd())
494
commands.append(command.DumpJSON())
477
496
if options.enable:
478
commands.append(EnableCmd())
497
commands.append(command.Enable())
480
499
if options.disable:
481
commands.append(DisableCmd())
500
commands.append(command.Disable())
483
502
if options.bump_timeout:
484
commands.append(BumpTimeoutCmd())
503
commands.append(command.BumpTimeout())
486
505
if options.start_checker:
487
commands.append(StartCheckerCmd())
506
commands.append(command.StartChecker())
489
508
if options.stop_checker:
490
commands.append(StopCheckerCmd())
509
commands.append(command.StopChecker())
492
511
if options.approved_by_default is not None:
493
512
if options.approved_by_default:
494
commands.append(ApproveByDefaultCmd())
513
commands.append(command.ApproveByDefault())
496
commands.append(DenyByDefaultCmd())
515
commands.append(command.DenyByDefault())
498
517
if options.checker is not None:
499
commands.append(SetCheckerCmd(options.checker))
518
commands.append(command.SetChecker(options.checker))
501
520
if options.host is not None:
502
commands.append(SetHostCmd(options.host))
521
commands.append(command.SetHost(options.host))
504
523
if options.secret is not None:
505
commands.append(SetSecretCmd(options.secret))
524
commands.append(command.SetSecret(options.secret))
507
526
if options.timeout is not None:
508
commands.append(SetTimeoutCmd(options.timeout))
527
commands.append(command.SetTimeout(options.timeout))
510
529
if options.extended_timeout:
512
SetExtendedTimeoutCmd(options.extended_timeout))
531
command.SetExtendedTimeout(options.extended_timeout))
514
533
if options.interval is not None:
515
commands.append(SetIntervalCmd(options.interval))
534
commands.append(command.SetInterval(options.interval))
517
536
if options.approval_delay is not None:
518
commands.append(SetApprovalDelayCmd(options.approval_delay))
538
command.SetApprovalDelay(options.approval_delay))
520
540
if options.approval_duration is not None:
522
SetApprovalDurationCmd(options.approval_duration))
542
command.SetApprovalDuration(options.approval_duration))
524
544
# If no command option has been given, show table of clients,
525
545
# optionally verbosely
527
commands.append(PrintTableCmd(verbose=options.verbose))
547
commands.append(command.PrintTable(verbose=options.verbose))
532
class Command(object):
533
"""Abstract class for commands"""
534
def run(self, clients, bus=None, mandos=None):
535
"""Normal commands should implement run_on_one_client(), but
536
commands which want to operate on all clients at the same time
537
can override this run() method instead."""
539
for clientpath, properties in clients.items():
540
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
541
dbus_busname, str(clientpath))
542
client = bus.get_object(dbus_busname, clientpath)
543
self.run_on_one_client(client, properties)
546
class IsEnabledCmd(Command):
547
def run(self, clients, bus=None, mandos=None):
548
client, properties = next(iter(clients.items()))
549
if self.is_enabled(client, properties):
552
def is_enabled(self, client, properties):
553
return properties["Enabled"]
556
class ApproveCmd(Command):
557
def run_on_one_client(self, client, properties):
558
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
559
client.__dbus_object_path__, client_dbus_interface)
560
client.Approve(dbus.Boolean(True),
561
dbus_interface=client_dbus_interface)
564
class DenyCmd(Command):
565
def run_on_one_client(self, client, properties):
566
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
567
client.__dbus_object_path__, client_dbus_interface)
568
client.Approve(dbus.Boolean(False),
569
dbus_interface=client_dbus_interface)
572
class RemoveCmd(Command):
573
def run_on_one_client(self, client, properties):
574
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
575
server_dbus_path, server_dbus_interface,
576
str(client.__dbus_object_path__))
577
self.mandos.RemoveClient(client.__dbus_object_path__)
580
class OutputCmd(Command):
581
"""Abstract class for commands outputting client details"""
582
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
583
"Created", "Interval", "Host", "KeyID",
584
"Fingerprint", "CheckerRunning", "LastEnabled",
585
"ApprovalPending", "ApprovedByDefault",
586
"LastApprovalRequest", "ApprovalDelay",
587
"ApprovalDuration", "Checker", "ExtendedTimeout",
588
"Expires", "LastCheckerStatus")
590
def run(self, clients, bus=None, mandos=None):
591
print(self.output(clients.values()))
593
def output(self, clients):
594
raise NotImplementedError()
597
class DumpJSONCmd(OutputCmd):
598
def output(self, clients):
599
data = {client["Name"]:
600
{key: self.dbus_boolean_to_bool(client[key])
601
for key in self.all_keywords}
602
for client in clients}
603
return json.dumps(data, indent=4, separators=(',', ': '))
606
def dbus_boolean_to_bool(value):
607
if isinstance(value, dbus.Boolean):
612
class PrintTableCmd(OutputCmd):
613
def __init__(self, verbose=False):
614
self.verbose = verbose
616
def output(self, clients):
617
default_keywords = ("Name", "Enabled", "Timeout",
619
keywords = default_keywords
621
keywords = self.all_keywords
622
return str(self.TableOfClients(clients, keywords))
624
class TableOfClients(object):
627
"Enabled": "Enabled",
628
"Timeout": "Timeout",
629
"LastCheckedOK": "Last Successful Check",
630
"LastApprovalRequest": "Last Approval Request",
631
"Created": "Created",
632
"Interval": "Interval",
634
"Fingerprint": "Fingerprint",
636
"CheckerRunning": "Check Is Running",
637
"LastEnabled": "Last Enabled",
638
"ApprovalPending": "Approval Is Pending",
639
"ApprovedByDefault": "Approved By Default",
640
"ApprovalDelay": "Approval Delay",
641
"ApprovalDuration": "Approval Duration",
642
"Checker": "Checker",
643
"ExtendedTimeout": "Extended Timeout",
644
"Expires": "Expires",
645
"LastCheckerStatus": "Last Checker Status",
648
def __init__(self, clients, keywords):
649
self.clients = clients
650
self.keywords = keywords
653
return "\n".join(self.rows())
655
if sys.version_info.major == 2:
656
__unicode__ = __str__
552
class command(object):
553
"""A namespace for command classes"""
556
"""Abstract base class for commands"""
557
def run(self, clients, bus=None, mandos=None):
558
"""Normal commands should implement run_on_one_client(),
559
but commands which want to operate on all clients at the same time can
560
override this run() method instead.
563
for clientpath, properties in clients.items():
564
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
565
dbus_busname, str(clientpath))
566
client = bus.get_object(dbus_busname, clientpath)
567
self.run_on_one_client(client, properties)
570
class IsEnabled(Base):
571
def run(self, clients, bus=None, mandos=None):
572
client, properties = next(iter(clients.items()))
573
if self.is_enabled(client, properties):
576
def is_enabled(self, client, properties):
577
return properties["Enabled"]
581
def run_on_one_client(self, client, properties):
582
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
583
client.__dbus_object_path__,
584
client_dbus_interface)
585
client.Approve(dbus.Boolean(True),
586
dbus_interface=client_dbus_interface)
590
def run_on_one_client(self, client, properties):
591
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
592
client.__dbus_object_path__,
593
client_dbus_interface)
594
client.Approve(dbus.Boolean(False),
595
dbus_interface=client_dbus_interface)
599
def run_on_one_client(self, client, properties):
600
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
601
dbus_busname, server_dbus_path,
602
server_dbus_interface,
603
str(client.__dbus_object_path__))
604
self.mandos.RemoveClient(client.__dbus_object_path__)
608
"""Abstract class for commands outputting client details"""
609
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
610
"Created", "Interval", "Host", "KeyID",
611
"Fingerprint", "CheckerRunning",
612
"LastEnabled", "ApprovalPending",
613
"ApprovedByDefault", "LastApprovalRequest",
614
"ApprovalDelay", "ApprovalDuration",
615
"Checker", "ExtendedTimeout", "Expires",
619
class DumpJSON(Output):
620
def run(self, clients, bus=None, mandos=None):
621
data = {client["Name"]:
622
{key: self.dbus_boolean_to_bool(client[key])
623
for key in self.all_keywords}
624
for client in clients.values()}
625
print(json.dumps(data, indent=4, separators=(',', ': ')))
628
def dbus_boolean_to_bool(value):
629
if isinstance(value, dbus.Boolean):
634
class PrintTable(Output):
635
def __init__(self, verbose=False):
636
self.verbose = verbose
638
def run(self, clients, bus=None, mandos=None):
639
default_keywords = ("Name", "Enabled", "Timeout",
641
keywords = default_keywords
643
keywords = self.all_keywords
644
print(self.TableOfClients(clients.values(), keywords))
646
class TableOfClients(object):
649
"Enabled": "Enabled",
650
"Timeout": "Timeout",
651
"LastCheckedOK": "Last Successful Check",
652
"LastApprovalRequest": "Last Approval Request",
653
"Created": "Created",
654
"Interval": "Interval",
656
"Fingerprint": "Fingerprint",
658
"CheckerRunning": "Check Is Running",
659
"LastEnabled": "Last Enabled",
660
"ApprovalPending": "Approval Is Pending",
661
"ApprovedByDefault": "Approved By Default",
662
"ApprovalDelay": "Approval Delay",
663
"ApprovalDuration": "Approval Duration",
664
"Checker": "Checker",
665
"ExtendedTimeout": "Extended Timeout",
666
"Expires": "Expires",
667
"LastCheckerStatus": "Last Checker Status",
670
def __init__(self, clients, keywords):
671
self.clients = clients
672
self.keywords = keywords
657
674
def __str__(self):
658
return str(self).encode(locale.getpreferredencoding())
661
format_string = self.row_formatting_string()
662
rows = [self.header_line(format_string)]
663
rows.extend(self.client_line(client, format_string)
664
for client in self.clients)
667
def row_formatting_string(self):
668
"Format string used to format table rows"
669
return " ".join("{{{key}:{width}}}".format(
670
width=max(len(self.tableheaders[key]),
671
*(len(self.string_from_client(client, key))
672
for client in self.clients)),
674
for key in self.keywords)
676
def string_from_client(self, client, key):
677
return self.valuetostring(client[key], key)
680
def valuetostring(cls, value, keyword):
681
if isinstance(value, dbus.Boolean):
682
return "Yes" if value else "No"
683
if keyword in ("Timeout", "Interval", "ApprovalDelay",
684
"ApprovalDuration", "ExtendedTimeout"):
685
return cls.milliseconds_to_string(value)
688
def header_line(self, format_string):
689
return format_string.format(**self.tableheaders)
691
def client_line(self, client, format_string):
692
return format_string.format(
693
**{key: self.string_from_client(client, key)
694
for key in self.keywords})
697
def milliseconds_to_string(ms):
698
td = datetime.timedelta(0, 0, 0, ms)
699
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
700
.format(days="{}T".format(td.days)
702
hours=td.seconds // 3600,
703
minutes=(td.seconds % 3600) // 60,
704
seconds=td.seconds % 60))
707
class PropertyCmd(Command):
708
"""Abstract class for Actions for setting one client property"""
710
def run_on_one_client(self, client, properties):
711
"""Set the Client's D-Bus property"""
712
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
713
client.__dbus_object_path__,
714
dbus.PROPERTIES_IFACE, client_dbus_interface,
715
self.propname, self.value_to_set
716
if not isinstance(self.value_to_set, dbus.Boolean)
717
else bool(self.value_to_set))
718
client.Set(client_dbus_interface, self.propname,
720
dbus_interface=dbus.PROPERTIES_IFACE)
724
raise NotImplementedError()
727
class EnableCmd(PropertyCmd):
729
value_to_set = dbus.Boolean(True)
732
class DisableCmd(PropertyCmd):
734
value_to_set = dbus.Boolean(False)
737
class BumpTimeoutCmd(PropertyCmd):
738
propname = "LastCheckedOK"
742
class StartCheckerCmd(PropertyCmd):
743
propname = "CheckerRunning"
744
value_to_set = dbus.Boolean(True)
747
class StopCheckerCmd(PropertyCmd):
748
propname = "CheckerRunning"
749
value_to_set = dbus.Boolean(False)
752
class ApproveByDefaultCmd(PropertyCmd):
753
propname = "ApprovedByDefault"
754
value_to_set = dbus.Boolean(True)
757
class DenyByDefaultCmd(PropertyCmd):
758
propname = "ApprovedByDefault"
759
value_to_set = dbus.Boolean(False)
762
class PropertyValueCmd(PropertyCmd):
763
"""Abstract class for PropertyCmd recieving a value as argument"""
764
def __init__(self, value):
765
self.value_to_set = value
768
class SetCheckerCmd(PropertyValueCmd):
772
class SetHostCmd(PropertyValueCmd):
776
class SetSecretCmd(PropertyValueCmd):
780
def value_to_set(self):
784
def value_to_set(self, value):
785
"""When setting, read data from supplied file object"""
786
self._vts = value.read()
790
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
791
"""Abstract class for PropertyValueCmd taking a value argument as
675
return "\n".join(self.rows())
677
if sys.version_info.major == 2:
678
__unicode__ = __str__
680
return str(self).encode(
681
locale.getpreferredencoding())
684
format_string = self.row_formatting_string()
685
rows = [self.header_line(format_string)]
686
rows.extend(self.client_line(client, format_string)
687
for client in self.clients)
690
def row_formatting_string(self):
691
"Format string used to format table rows"
692
return " ".join("{{{key}:{width}}}".format(
693
width=max(len(self.tableheaders[key]),
694
*(len(self.string_from_client(client,
696
for client in self.clients)),
698
for key in self.keywords)
700
def string_from_client(self, client, key):
701
return self.valuetostring(client[key], key)
704
def valuetostring(cls, value, keyword):
705
if isinstance(value, dbus.Boolean):
706
return "Yes" if value else "No"
707
if keyword in ("Timeout", "Interval", "ApprovalDelay",
708
"ApprovalDuration", "ExtendedTimeout"):
709
return cls.milliseconds_to_string(value)
712
def header_line(self, format_string):
713
return format_string.format(**self.tableheaders)
715
def client_line(self, client, format_string):
716
return format_string.format(
717
**{key: self.string_from_client(client, key)
718
for key in self.keywords})
721
def milliseconds_to_string(ms):
722
td = datetime.timedelta(0, 0, 0, ms)
723
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
724
.format(days="{}T".format(td.days)
726
hours=td.seconds // 3600,
727
minutes=(td.seconds % 3600) // 60,
728
seconds=td.seconds % 60))
731
class Property(Base):
732
"Abstract class for Actions for setting one client property"
734
def run_on_one_client(self, client, properties):
735
"""Set the Client's D-Bus property"""
736
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
737
client.__dbus_object_path__,
738
dbus.PROPERTIES_IFACE, client_dbus_interface,
739
self.propname, self.value_to_set
740
if not isinstance(self.value_to_set,
742
else bool(self.value_to_set))
743
client.Set(client_dbus_interface, self.propname,
745
dbus_interface=dbus.PROPERTIES_IFACE)
749
raise NotImplementedError()
752
class Enable(Property):
754
value_to_set = dbus.Boolean(True)
757
class Disable(Property):
759
value_to_set = dbus.Boolean(False)
762
class BumpTimeout(Property):
763
propname = "LastCheckedOK"
767
class StartChecker(Property):
768
propname = "CheckerRunning"
769
value_to_set = dbus.Boolean(True)
772
class StopChecker(Property):
773
propname = "CheckerRunning"
774
value_to_set = dbus.Boolean(False)
777
class ApproveByDefault(Property):
778
propname = "ApprovedByDefault"
779
value_to_set = dbus.Boolean(True)
782
class DenyByDefault(Property):
783
propname = "ApprovedByDefault"
784
value_to_set = dbus.Boolean(False)
787
class PropertyValue(Property):
788
"Abstract class for Property recieving a value as argument"
789
def __init__(self, value):
790
self.value_to_set = value
793
class SetChecker(PropertyValue):
797
class SetHost(PropertyValue):
801
class SetSecret(PropertyValue):
805
def value_to_set(self):
809
def value_to_set(self, value):
810
"""When setting, read data from supplied file object"""
811
self._vts = value.read()
815
class MillisecondsPropertyValueArgument(PropertyValue):
816
"""Abstract class for PropertyValue taking a value argument as
792
817
a datetime.timedelta() but should store it as milliseconds."""
795
def value_to_set(self):
799
def value_to_set(self, value):
800
"""When setting, convert value from a datetime.timedelta"""
801
self._vts = int(round(value.total_seconds() * 1000))
804
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
808
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
809
propname = "ExtendedTimeout"
812
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
813
propname = "Interval"
816
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
817
propname = "ApprovalDelay"
820
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
821
propname = "ApprovalDuration"
820
def value_to_set(self):
824
def value_to_set(self, value):
825
"When setting, convert value from a datetime.timedelta"
826
self._vts = int(round(value.total_seconds() * 1000))
829
class SetTimeout(MillisecondsPropertyValueArgument):
833
class SetExtendedTimeout(MillisecondsPropertyValueArgument):
834
propname = "ExtendedTimeout"
837
class SetInterval(MillisecondsPropertyValueArgument):
838
propname = "Interval"
841
class SetApprovalDelay(MillisecondsPropertyValueArgument):
842
propname = "ApprovalDelay"
845
class SetApprovalDuration(MillisecondsPropertyValueArgument):
846
propname = "ApprovalDuration"
825
class Test_string_to_delta(unittest.TestCase):
826
def test_handles_basic_rfc3339(self):
827
self.assertEqual(string_to_delta("PT0S"),
828
datetime.timedelta())
829
self.assertEqual(string_to_delta("P0D"),
830
datetime.timedelta())
831
self.assertEqual(string_to_delta("PT1S"),
832
datetime.timedelta(0, 1))
833
self.assertEqual(string_to_delta("PT2H"),
834
datetime.timedelta(0, 7200))
850
class TestCaseWithAssertLogs(unittest.TestCase):
851
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
853
if not hasattr(unittest.TestCase, "assertLogs"):
854
@contextlib.contextmanager
855
def assertLogs(self, logger, level=logging.INFO):
856
capturing_handler = self.CapturingLevelHandler(level)
857
old_level = logger.level
858
old_propagate = logger.propagate
859
logger.addHandler(capturing_handler)
860
logger.setLevel(level)
861
logger.propagate = False
863
yield capturing_handler.watcher
865
logger.propagate = old_propagate
866
logger.removeHandler(capturing_handler)
867
logger.setLevel(old_level)
868
self.assertGreater(len(capturing_handler.watcher.records),
871
class CapturingLevelHandler(logging.Handler):
872
def __init__(self, level, *args, **kwargs):
873
logging.Handler.__init__(self, *args, **kwargs)
874
self.watcher = self.LoggingWatcher([], [])
875
def emit(self, record):
876
self.watcher.records.append(record)
877
self.watcher.output.append(self.format(record))
879
LoggingWatcher = collections.namedtuple("LoggingWatcher",
884
class Test_string_to_delta(TestCaseWithAssertLogs):
885
# Just test basic RFC 3339 functionality here, the doc string for
886
# rfc3339_duration_to_delta() already has more comprehensive
887
# tests, which is run by doctest.
889
def test_rfc3339_zero_seconds(self):
890
self.assertEqual(datetime.timedelta(),
891
string_to_delta("PT0S"))
893
def test_rfc3339_zero_days(self):
894
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
896
def test_rfc3339_one_second(self):
897
self.assertEqual(datetime.timedelta(0, 1),
898
string_to_delta("PT1S"))
900
def test_rfc3339_two_hours(self):
901
self.assertEqual(datetime.timedelta(0, 7200),
902
string_to_delta("PT2H"))
836
904
def test_falls_back_to_pre_1_6_1_with_warning(self):
837
# assertLogs only exists in Python 3.4
838
if hasattr(self, "assertLogs"):
839
with self.assertLogs(log, logging.WARNING):
840
value = string_to_delta("2h")
842
class WarningFilter(logging.Filter):
843
"""Don't show, but record the presence of, warnings"""
844
def filter(self, record):
845
is_warning = record.levelno >= logging.WARNING
846
self.found = is_warning or getattr(self, "found",
848
return not is_warning
849
warning_filter = WarningFilter()
850
log.addFilter(warning_filter)
852
value = string_to_delta("2h")
854
log.removeFilter(warning_filter)
855
self.assertTrue(getattr(warning_filter, "found", False))
856
self.assertEqual(value, datetime.timedelta(0, 7200))
905
with self.assertLogs(log, logging.WARNING):
906
value = string_to_delta("2h")
907
self.assertEqual(datetime.timedelta(0, 7200), value)
859
910
class Test_check_option_syntax(unittest.TestCase):
1032
1167
options = self.parser.parse_args(args)
1033
1168
check_option_syntax(self.parser, options)
1034
1169
commands = commands_from_options(options)
1035
self.assertEqual(len(commands), 1)
1170
self.assertEqual(1, len(commands))
1036
1171
command = commands[0]
1037
1172
self.assertIsInstance(command, command_cls)
1038
1173
for key, value in cmd_attrs.items():
1039
self.assertEqual(getattr(command, key), value)
1174
self.assertEqual(value, getattr(command, key))
1041
1176
def test_is_enabled_short(self):
1042
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1177
self.assert_command_from_args(["-V", "foo"],
1044
1180
def test_approve(self):
1045
1181
self.assert_command_from_args(["--approve", "foo"],
1048
1184
def test_approve_short(self):
1049
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1185
self.assert_command_from_args(["-A", "foo"], command.Approve)
1051
1187
def test_deny(self):
1052
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1188
self.assert_command_from_args(["--deny", "foo"], command.Deny)
1054
1190
def test_deny_short(self):
1055
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1191
self.assert_command_from_args(["-D", "foo"], command.Deny)
1057
1193
def test_remove(self):
1058
1194
self.assert_command_from_args(["--remove", "foo"],
1061
1197
def test_deny_before_remove(self):
1062
1198
options = self.parser.parse_args(["--deny", "--remove",
1064
1200
check_option_syntax(self.parser, options)
1065
1201
commands = commands_from_options(options)
1066
self.assertEqual(len(commands), 2)
1067
self.assertIsInstance(commands[0], DenyCmd)
1068
self.assertIsInstance(commands[1], RemoveCmd)
1202
self.assertEqual(2, len(commands))
1203
self.assertIsInstance(commands[0], command.Deny)
1204
self.assertIsInstance(commands[1], command.Remove)
1070
1206
def test_deny_before_remove_reversed(self):
1071
1207
options = self.parser.parse_args(["--remove", "--deny",
1073
1209
check_option_syntax(self.parser, options)
1074
1210
commands = commands_from_options(options)
1075
self.assertEqual(len(commands), 2)
1076
self.assertIsInstance(commands[0], DenyCmd)
1077
self.assertIsInstance(commands[1], RemoveCmd)
1211
self.assertEqual(2, len(commands))
1212
self.assertIsInstance(commands[0], command.Deny)
1213
self.assertIsInstance(commands[1], command.Remove)
1079
1215
def test_remove_short(self):
1080
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1216
self.assert_command_from_args(["-r", "foo"], command.Remove)
1082
1218
def test_dump_json(self):
1083
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1219
self.assert_command_from_args(["--dump-json"],
1085
1222
def test_enable(self):
1086
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1223
self.assert_command_from_args(["--enable", "foo"],
1088
1226
def test_enable_short(self):
1089
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1227
self.assert_command_from_args(["-e", "foo"], command.Enable)
1091
1229
def test_disable(self):
1092
1230
self.assert_command_from_args(["--disable", "foo"],
1095
1233
def test_disable_short(self):
1096
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1234
self.assert_command_from_args(["-d", "foo"], command.Disable)
1098
1236
def test_bump_timeout(self):
1099
1237
self.assert_command_from_args(["--bump-timeout", "foo"],
1238
command.BumpTimeout)
1102
1240
def test_bump_timeout_short(self):
1103
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1241
self.assert_command_from_args(["-b", "foo"],
1242
command.BumpTimeout)
1105
1244
def test_start_checker(self):
1106
1245
self.assert_command_from_args(["--start-checker", "foo"],
1246
command.StartChecker)
1109
1248
def test_stop_checker(self):
1110
1249
self.assert_command_from_args(["--stop-checker", "foo"],
1250
command.StopChecker)
1113
1252
def test_approve_by_default(self):
1114
1253
self.assert_command_from_args(["--approve-by-default", "foo"],
1115
ApproveByDefaultCmd)
1254
command.ApproveByDefault)
1117
1256
def test_deny_by_default(self):
1118
1257
self.assert_command_from_args(["--deny-by-default", "foo"],
1258
command.DenyByDefault)
1121
1260
def test_checker(self):
1122
1261
self.assert_command_from_args(["--checker", ":", "foo"],
1123
SetCheckerCmd, value_to_set=":")
1125
1265
def test_checker_empty(self):
1126
1266
self.assert_command_from_args(["--checker", "", "foo"],
1127
SetCheckerCmd, value_to_set="")
1129
1270
def test_checker_short(self):
1130
1271
self.assert_command_from_args(["-c", ":", "foo"],
1131
SetCheckerCmd, value_to_set=":")
1133
1275
def test_host(self):
1134
1276
self.assert_command_from_args(["--host", "foo.example.org",
1277
"foo"], command.SetHost,
1136
1278
value_to_set="foo.example.org")
1138
1280
def test_host_short(self):
1139
1281
self.assert_command_from_args(["-H", "foo.example.org",
1282
"foo"], command.SetHost,
1141
1283
value_to_set="foo.example.org")
1143
1285
def test_secret_devnull(self):
1144
1286
self.assert_command_from_args(["--secret", os.path.devnull,
1145
"foo"], SetSecretCmd,
1287
"foo"], command.SetSecret,
1146
1288
value_to_set=b"")
1148
1290
def test_secret_tempfile(self):
1315
class TestIsEnabledCmd(TestCmd):
1316
def test_is_enabled(self):
1317
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1319
for client, properties
1320
in self.clients.items()))
1456
class TestBaseCommands(TestCommand):
1322
def test_is_enabled_run_exits_successfully(self):
1458
def test_IsEnabled_exits_successfully(self):
1323
1459
with self.assertRaises(SystemExit) as e:
1324
IsEnabledCmd().run(self.one_client)
1460
command.IsEnabled().run(self.one_client)
1325
1461
if e.exception.code is not None:
1326
self.assertEqual(e.exception.code, 0)
1462
self.assertEqual(0, e.exception.code)
1328
1464
self.assertIsNone(e.exception.code)
1330
def test_is_enabled_run_exits_with_failure(self):
1466
def test_IsEnabled_exits_with_failure(self):
1331
1467
self.client.attributes["Enabled"] = dbus.Boolean(False)
1332
1468
with self.assertRaises(SystemExit) as e:
1333
IsEnabledCmd().run(self.one_client)
1469
command.IsEnabled().run(self.one_client)
1334
1470
if isinstance(e.exception.code, int):
1335
self.assertNotEqual(e.exception.code, 0)
1471
self.assertNotEqual(0, e.exception.code)
1337
1473
self.assertIsNotNone(e.exception.code)
1340
class TestApproveCmd(TestCmd):
1341
def test_approve(self):
1342
ApproveCmd().run(self.clients, self.bus)
1475
def test_Approve(self):
1476
command.Approve().run(self.clients, self.bus)
1343
1477
for clientpath in self.clients:
1344
1478
client = self.bus.get_object(dbus_busname, clientpath)
1345
1479
self.assertIn(("Approve", (True, client_dbus_interface)),
1349
class TestDenyCmd(TestCmd):
1350
def test_deny(self):
1351
DenyCmd().run(self.clients, self.bus)
1482
def test_Deny(self):
1483
command.Deny().run(self.clients, self.bus)
1352
1484
for clientpath in self.clients:
1353
1485
client = self.bus.get_object(dbus_busname, clientpath)
1354
1486
self.assertIn(("Approve", (False, client_dbus_interface)),
1358
class TestRemoveCmd(TestCmd):
1359
def test_remove(self):
1489
def test_Remove(self):
1360
1490
class MockMandos(object):
1361
1491
def __init__(self):
1362
1492
self.calls = []
1363
1493
def RemoveClient(self, dbus_path):
1364
1494
self.calls.append(("RemoveClient", (dbus_path,)))
1365
1495
mandos = MockMandos()
1366
super(TestRemoveCmd, self).setUp()
1367
RemoveCmd().run(self.clients, self.bus, mandos)
1368
self.assertEqual(len(mandos.calls), 2)
1496
command.Remove().run(self.clients, self.bus, mandos)
1369
1497
for clientpath in self.clients:
1370
1498
self.assertIn(("RemoveClient", (clientpath,)),
1374
class TestDumpJSONCmd(TestCmd):
1376
self.expected_json = {
1379
"KeyID": ("92ed150794387c03ce684574b1139a65"
1380
"94a34f895daaaf09fd8ea90a27cddb12"),
1381
"Host": "foo.example.org",
1384
"LastCheckedOK": "2019-02-03T00:00:00",
1385
"Created": "2019-01-02T00:00:00",
1387
"Fingerprint": ("778827225BA7DE539C5A"
1388
"7CFA59CFF7CDBD9A5920"),
1389
"CheckerRunning": False,
1390
"LastEnabled": "2019-01-03T00:00:00",
1391
"ApprovalPending": False,
1392
"ApprovedByDefault": True,
1393
"LastApprovalRequest": "",
1395
"ApprovalDuration": 1000,
1396
"Checker": "fping -q -- %(host)s",
1397
"ExtendedTimeout": 900000,
1398
"Expires": "2019-02-04T00:00:00",
1399
"LastCheckerStatus": 0,
1403
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1404
"6ab612cff5ad227247e46c2b020f441c"),
1405
"Host": "192.0.2.3",
1408
"LastCheckedOK": "2019-02-04T00:00:00",
1409
"Created": "2019-01-03T00:00:00",
1411
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1412
"F547B3A107558FCA3A27"),
1413
"CheckerRunning": True,
1414
"LastEnabled": "2019-01-04T00:00:00",
1415
"ApprovalPending": False,
1416
"ApprovedByDefault": False,
1417
"LastApprovalRequest": "2019-01-03T00:00:00",
1418
"ApprovalDelay": 30000,
1419
"ApprovalDuration": 93785000,
1421
"ExtendedTimeout": 900000,
1422
"Expires": "2019-02-05T00:00:00",
1423
"LastCheckerStatus": -2,
1426
return super(TestDumpJSONCmd, self).setUp()
1428
def test_normal(self):
1429
output = DumpJSONCmd().output(self.clients.values())
1430
json_data = json.loads(output)
1431
self.assertDictEqual(json_data, self.expected_json)
1433
def test_one_client(self):
1434
output = DumpJSONCmd().output(self.one_client.values())
1435
json_data = json.loads(output)
1504
"KeyID": ("92ed150794387c03ce684574b1139a65"
1505
"94a34f895daaaf09fd8ea90a27cddb12"),
1506
"Host": "foo.example.org",
1509
"LastCheckedOK": "2019-02-03T00:00:00",
1510
"Created": "2019-01-02T00:00:00",
1512
"Fingerprint": ("778827225BA7DE539C5A"
1513
"7CFA59CFF7CDBD9A5920"),
1514
"CheckerRunning": False,
1515
"LastEnabled": "2019-01-03T00:00:00",
1516
"ApprovalPending": False,
1517
"ApprovedByDefault": True,
1518
"LastApprovalRequest": "",
1520
"ApprovalDuration": 1000,
1521
"Checker": "fping -q -- %(host)s",
1522
"ExtendedTimeout": 900000,
1523
"Expires": "2019-02-04T00:00:00",
1524
"LastCheckerStatus": 0,
1528
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1529
"6ab612cff5ad227247e46c2b020f441c"),
1530
"Host": "192.0.2.3",
1533
"LastCheckedOK": "2019-02-04T00:00:00",
1534
"Created": "2019-01-03T00:00:00",
1536
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1537
"F547B3A107558FCA3A27"),
1538
"CheckerRunning": True,
1539
"LastEnabled": "2019-01-04T00:00:00",
1540
"ApprovalPending": False,
1541
"ApprovedByDefault": False,
1542
"LastApprovalRequest": "2019-01-03T00:00:00",
1543
"ApprovalDelay": 30000,
1544
"ApprovalDuration": 93785000,
1546
"ExtendedTimeout": 900000,
1547
"Expires": "2019-02-05T00:00:00",
1548
"LastCheckerStatus": -2,
1552
def test_DumpJSON_normal(self):
1553
with self.capture_stdout_to_buffer() as buffer:
1554
command.DumpJSON().run(self.clients)
1555
json_data = json.loads(buffer.getvalue())
1556
self.assertDictEqual(self.expected_json, json_data)
1559
@contextlib.contextmanager
1560
def capture_stdout_to_buffer():
1561
capture_buffer = io.StringIO()
1562
old_stdout = sys.stdout
1563
sys.stdout = capture_buffer
1565
yield capture_buffer
1567
sys.stdout = old_stdout
1569
def test_DumpJSON_one_client(self):
1570
with self.capture_stdout_to_buffer() as buffer:
1571
command.DumpJSON().run(self.one_client)
1572
json_data = json.loads(buffer.getvalue())
1436
1573
expected_json = {"foo": self.expected_json["foo"]}
1437
self.assertDictEqual(json_data, expected_json)
1440
class TestPrintTableCmd(TestCmd):
1441
def test_normal(self):
1442
output = PrintTableCmd().output(self.clients.values())
1574
self.assertDictEqual(expected_json, json_data)
1576
def test_PrintTable_normal(self):
1577
with self.capture_stdout_to_buffer() as buffer:
1578
command.PrintTable().run(self.clients)
1443
1579
expected_output = "\n".join((
1444
1580
"Name Enabled Timeout Last Successful Check",
1445
1581
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1446
1582
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1448
self.assertEqual(output, expected_output)
1584
self.assertEqual(expected_output, buffer.getvalue())
1450
def test_verbose(self):
1451
output = PrintTableCmd(verbose=True).output(
1452
self.clients.values())
1586
def test_PrintTable_verbose(self):
1587
with self.capture_stdout_to_buffer() as buffer:
1588
command.PrintTable(verbose=True).run(self.clients)