To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 1017
- compare with another revision
- download diff
- download tarball
- view history from revision 1017
- Committer: Teddy Hogeborn
- Date: 2019-03-04 18:43:22 UTC
- Revision ID: teddy@recompile.se-20190304184322-6tumlykqh57t5nvs
mandos-ctl: Add test for RemoveCmd
* mandos-ctl (TestRemoveCmd): New.
* mandos-ctl (TestRemoveCmd): New.
- files modified:
- mandos-ctl
added
removed
mandos-ctl
42
42
import json
43
43
import unittest
44
44
import logging
45
import io
46
import tempfile
47
import contextlib
48
45
49
46
import dbus
50
47
281
278
commands which want to operate on all clients at the same time
282
279
can override this run() method instead."""
283
280
self.mandos = mandos
284
for client, properties in clients.items():
285
self.run_on_one_client(client, properties)
281
for client in clients:
282
self.run_on_one_client(client)
286
283
287
284
class PrintCmd(Command):
288
285
"""Abstract class for commands printing client details"""
298
295
299
296
class PropertyCmd(Command):
300
297
"""Abstract class for Actions for setting one client property"""
301
def run_on_one_client(self, client, properties):
298
def run_on_one_client(self, client):
302
299
"""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
300
client.Set(client_interface, self.property, self.value_to_set,
310
301
dbus_interface=dbus.PROPERTIES_IFACE)
311
302
323
314
@value_to_set.setter
324
315
def value_to_set(self, value):
325
316
"""When setting, convert value to a datetime.timedelta"""
326
self._vts = int(round(value.total_seconds() * 1000))
317
self._vts = string_to_delta(value).total_seconds() * 1000
327
318
328
319
# Actual (non-abstract) command classes
329
320
332
323
self.verbose = verbose
333
324
334
325
def output(self, clients):
335
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
336
keywords = default_keywords
337
326
if self.verbose:
338
327
keywords = self.all_keywords
328
else:
329
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
339
330
return str(self.TableOfClients(clients.values(), keywords))
340
331
341
332
class TableOfClients(object):
428
419
return value
429
420
430
421
class IsEnabledCmd(Command):
431
def run_on_one_client(self, client, properties):
432
if self.is_enabled(client, properties):
422
def run_on_one_client(self, client):
423
if self.is_enabled(client):
433
424
sys.exit(0)
434
425
sys.exit(1)
435
def is_enabled(self, client, properties):
436
return bool(properties["Enabled"])
426
def is_enabled(self, client):
427
return client.Get(client_interface, "Enabled",
428
dbus_interface=dbus.PROPERTIES_IFACE)
437
429
438
430
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__))
431
def run_on_one_client(self, client):
443
432
self.mandos.RemoveClient(client.__dbus_object_path__)
444
433
445
434
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)
435
def run_on_one_client(self, client):
449
436
client.Approve(dbus.Boolean(True),
450
437
dbus_interface=client_interface)
451
438
452
439
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)
440
def run_on_one_client(self, client):
456
441
client.Approve(dbus.Boolean(False),
457
442
dbus_interface=client_interface)
458
443
491
476
property = "Host"
492
477
493
478
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
494
@property
495
def value_to_set(self):
496
return self._vts
497
@value_to_set.setter
498
def value_to_set(self, value):
499
"""When setting, read data from supplied file object"""
500
self._vts = value.read()
501
value.close()
502
479
property = "Secret"
503
480
504
481
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
519
496
MillisecondsValueArgumentMixIn):
520
497
property = "ApprovalDuration"
521
498
499
def has_actions(options):
500
return any((options.enable,
501
options.disable,
502
options.bump_timeout,
503
options.start_checker,
504
options.stop_checker,
505
options.is_enabled,
506
options.remove,
507
options.checker is not None,
508
options.timeout is not None,
509
options.extended_timeout is not None,
510
options.interval is not None,
511
options.approved_by_default is not None,
512
options.approval_delay is not None,
513
options.approval_duration is not None,
514
options.host is not None,
515
options.secret is not None,
516
options.approve,
517
options.deny))
518
522
519
def add_command_line_options(parser):
523
520
parser.add_argument("--version", action="version",
524
521
version="%(prog)s {}".format(version),
550
547
help="Remove client")
551
548
parser.add_argument("-c", "--checker",
552
549
help="Set checker command for client")
553
parser.add_argument("-t", "--timeout", type=string_to_delta,
550
parser.add_argument("-t", "--timeout",
554
551
help="Set timeout for client")
555
parser.add_argument("--extended-timeout", type=string_to_delta,
552
parser.add_argument("--extended-timeout",
556
553
help="Set extended timeout for client")
557
parser.add_argument("-i", "--interval", type=string_to_delta,
554
parser.add_argument("-i", "--interval",
558
555
help="Set checker interval for client")
559
556
approve_deny_default = parser.add_mutually_exclusive_group()
560
557
approve_deny_default.add_argument(
565
562
"--deny-by-default", action="store_false",
566
563
dest="approved_by_default",
567
564
help="Set client to be denied by default")
568
parser.add_argument("--approval-delay", type=string_to_delta,
565
parser.add_argument("--approval-delay",
569
566
help="Set delay before client approve/deny")
570
parser.add_argument("--approval-duration", type=string_to_delta,
567
parser.add_argument("--approval-duration",
571
568
help="Set duration of one client approval")
572
569
parser.add_argument("-H", "--host", help="Set host for client")
573
570
parser.add_argument("-s", "--secret",
579
576
help="Approve any current client request")
580
577
approve_deny.add_argument("-D", "--deny", action="store_true",
581
578
help="Deny any current client request")
582
parser.add_argument("--debug", action="store_true",
583
help="Debug mode (show D-Bus commands)")
584
579
parser.add_argument("--check", action="store_true",
585
580
help="Run self-test")
586
581
parser.add_argument("client", nargs="*", help="Client name")
587
582
588
583
589
def commands_from_options(options):
584
def commands_and_clients_from_options(options):
590
585
591
586
commands = []
592
587
600
595
commands.append(DisableCmd())
601
596
602
597
if options.bump_timeout:
603
commands.append(BumpTimeoutCmd())
598
commands.append(BumpTimeoutCmd(options.bump_timeout))
604
599
605
600
if options.start_checker:
606
601
commands.append(StartCheckerCmd())
615
610
commands.append(RemoveCmd())
616
611
617
612
if options.checker is not None:
618
commands.append(SetCheckerCmd(options.checker))
613
commands.append(SetCheckerCmd())
619
614
620
615
if options.timeout is not None:
621
616
commands.append(SetTimeoutCmd(options.timeout))
625
620
SetExtendedTimeoutCmd(options.extended_timeout))
626
621
627
622
if options.interval is not None:
628
commands.append(SetIntervalCmd(options.interval))
623
command.append(SetIntervalCmd(options.interval))
629
624
630
625
if options.approved_by_default is not None:
631
626
if options.approved_by_default:
632
commands.append(ApproveByDefaultCmd())
627
command.append(ApproveByDefaultCmd())
633
628
else:
634
commands.append(DenyByDefaultCmd())
629
command.append(DenyByDefaultCmd())
635
630
636
631
if options.approval_delay is not None:
637
commands.append(SetApprovalDelayCmd(options.approval_delay))
632
command.append(SetApprovalDelayCmd(options.approval_delay))
638
633
639
634
if options.approval_duration is not None:
640
commands.append(
635
command.append(
641
636
SetApprovalDurationCmd(options.approval_duration))
642
637
643
638
if options.host is not None:
644
commands.append(SetHostCmd(options.host))
639
command.append(SetHostCmd(options.host))
645
640
646
641
if options.secret is not None:
647
commands.append(SetSecretCmd(options.secret))
642
command.append(SetSecretCmd(options.secret))
648
643
649
644
if options.approve:
650
645
commands.append(ApproveCmd())
657
652
if not commands:
658
653
commands.append(PrintTableCmd(verbose=options.verbose))
659
654
660
return commands
661
662
663
def check_option_syntax(parser, options):
664
"""Apply additional restrictions on options, not expressible in
665
argparse"""
666
667
def has_actions(options):
668
return any((options.enable,
669
options.disable,
670
options.bump_timeout,
671
options.start_checker,
672
options.stop_checker,
673
options.is_enabled,
674
options.remove,
675
options.checker is not None,
676
options.timeout is not None,
677
options.extended_timeout is not None,
678
options.interval is not None,
679
options.approved_by_default is not None,
680
options.approval_delay is not None,
681
options.approval_duration is not None,
682
options.host is not None,
683
options.secret is not None,
684
options.approve,
685
options.deny))
655
return commands, options.client
656
657
658
def main():
659
parser = argparse.ArgumentParser()
660
661
add_command_line_options(parser)
662
663
options = parser.parse_args()
686
664
687
665
if has_actions(options) and not (options.client or options.all):
688
666
parser.error("Options require clients names or --all.")
696
674
if options.is_enabled and len(options.client) > 1:
697
675
parser.error("--is-enabled requires exactly one client")
698
676
699
700
def main():
701
parser = argparse.ArgumentParser()
702
703
add_command_line_options(parser)
704
705
options = parser.parse_args()
706
707
check_option_syntax(parser, options)
708
709
clientnames = options.client
710
711
if options.debug:
712
log.setLevel(logging.DEBUG)
677
commands, clientnames = commands_and_clients_from_options(options)
713
678
714
679
try:
715
680
bus = dbus.SystemBus()
716
log.debug("D-Bus: Connect to: (name=%r, path=%r)", busname,
717
server_path)
718
681
mandos_dbus_objc = bus.get_object(busname, server_path)
719
682
except dbus.exceptions.DBusException:
720
683
log.critical("Could not connect to Mandos server")
733
696
dbus_filter = NullFilter()
734
697
try:
735
698
dbus_logger.addFilter(dbus_filter)
736
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,
737
server_path, dbus.OBJECT_MANAGER_IFACE)
738
699
mandos_clients = {path: ifs_and_props[client_interface]
739
700
for path, ifs_and_props in
740
701
mandos_serv_object_manager
766
727
sys.exit(1)
767
728
768
729
# Run all commands on clients
769
commands = commands_from_options(options)
770
730
for command in commands:
771
731
command.run(mandos_serv, clients)
772
732
786
746
787
747
class Test_string_to_delta(unittest.TestCase):
788
748
def test_handles_basic_rfc3339(self):
789
self.assertEqual(string_to_delta("PT0S"),
790
datetime.timedelta())
791
self.assertEqual(string_to_delta("P0D"),
792
datetime.timedelta())
793
self.assertEqual(string_to_delta("PT1S"),
794
datetime.timedelta(0, 1))
795
749
self.assertEqual(string_to_delta("PT2H"),
796
750
datetime.timedelta(0, 7200))
797
751
def test_falls_back_to_pre_1_6_1_with_warning(self):
832
786
testcase.assertEqual(dbus_interface,
833
787
dbus.PROPERTIES_IFACE)
834
788
self.attributes[property] = value
789
self.calls.append(("Set", (interface, property, value,
790
dbus_interface)))
835
791
def Get(self, interface, property, dbus_interface):
836
792
testcase.assertEqual(interface, client_interface)
837
793
testcase.assertEqual(dbus_interface,
838
794
dbus.PROPERTIES_IFACE)
795
self.calls.append(("Get", (interface, property,
796
dbus_interface)))
839
797
return self.attributes[property]
840
def Approve(self, approve, dbus_interface):
841
testcase.assertEqual(dbus_interface, client_interface)
842
self.calls.append(("Approve", (approve,
843
dbus_interface)))
844
self.client = MockClient(
845
"foo",
846
KeyID=("92ed150794387c03ce684574b1139a65"
847
"94a34f895daaaf09fd8ea90a27cddb12"),
848
Secret=b"secret",
849
Host="foo.example.org",
850
Enabled=dbus.Boolean(True),
851
Timeout=300000,
852
LastCheckedOK="2019-02-03T00:00:00",
853
Created="2019-01-02T00:00:00",
854
Interval=120000,
855
Fingerprint=("778827225BA7DE539C5A"
856
"7CFA59CFF7CDBD9A5920"),
857
CheckerRunning=dbus.Boolean(False),
858
LastEnabled="2019-01-03T00:00:00",
859
ApprovalPending=dbus.Boolean(False),
860
ApprovedByDefault=dbus.Boolean(True),
861
LastApprovalRequest="",
862
ApprovalDelay=0,
863
ApprovalDuration=1000,
864
Checker="fping -q -- %(host)s",
865
ExtendedTimeout=900000,
866
Expires="2019-02-04T00:00:00",
867
LastCheckerStatus=0)
868
self.other_client = MockClient(
869
"barbar",
870
KeyID=("0558568eedd67d622f5c83b35a115f79"
871
"6ab612cff5ad227247e46c2b020f441c"),
872
Secret=b"secretbar",
873
Host="192.0.2.3",
874
Enabled=dbus.Boolean(True),
875
Timeout=300000,
876
LastCheckedOK="2019-02-04T00:00:00",
877
Created="2019-01-03T00:00:00",
878
Interval=120000,
879
Fingerprint=("3E393AEAEFB84C7E89E2"
880
"F547B3A107558FCA3A27"),
881
CheckerRunning=dbus.Boolean(True),
882
LastEnabled="2019-01-04T00:00:00",
883
ApprovalPending=dbus.Boolean(False),
884
ApprovedByDefault=dbus.Boolean(False),
885
LastApprovalRequest="2019-01-03T00:00:00",
886
ApprovalDelay=30000,
887
ApprovalDuration=1000,
888
Checker=":",
889
ExtendedTimeout=900000,
890
Expires="2019-02-05T00:00:00",
891
LastCheckerStatus=-2)
892
self.clients = collections.OrderedDict(
893
[
894
(self.client, self.client.attributes),
895
(self.other_client, self.other_client.attributes),
798
def __getitem__(self, key):
799
return self.attributes[key]
800
def __setitem__(self, key, value):
801
self.attributes[key] = value
802
self.clients = collections.OrderedDict([
803
("foo",
804
MockClient(
805
"foo",
806
KeyID=("92ed150794387c03ce684574b1139a65"
807
"94a34f895daaaf09fd8ea90a27cddb12"),
808
Secret=b"secret",
809
Host="foo.example.org",
810
Enabled=dbus.Boolean(True),
811
Timeout=300000,
812
LastCheckedOK="2019-02-03T00:00:00",
813
Created="2019-01-02T00:00:00",
814
Interval=120000,
815
Fingerprint=("778827225BA7DE539C5A"
816
"7CFA59CFF7CDBD9A5920"),
817
CheckerRunning=dbus.Boolean(False),
818
LastEnabled="2019-01-03T00:00:00",
819
ApprovalPending=dbus.Boolean(False),
820
ApprovedByDefault=dbus.Boolean(True),
821
LastApprovalRequest="",
822
ApprovalDelay=0,
823
ApprovalDuration=1000,
824
Checker="fping -q -- %(host)s",
825
ExtendedTimeout=900000,
826
Expires="2019-02-04T00:00:00",
827
LastCheckerStatus=0)),
828
("barbar",
829
MockClient(
830
"barbar",
831
KeyID=("0558568eedd67d622f5c83b35a115f79"
832
"6ab612cff5ad227247e46c2b020f441c"),
833
Secret=b"secretbar",
834
Host="192.0.2.3",
835
Enabled=dbus.Boolean(True),
836
Timeout=300000,
837
LastCheckedOK="2019-02-04T00:00:00",
838
Created="2019-01-03T00:00:00",
839
Interval=120000,
840
Fingerprint=("3E393AEAEFB84C7E89E2"
841
"F547B3A107558FCA3A27"),
842
CheckerRunning=dbus.Boolean(True),
843
LastEnabled="2019-01-04T00:00:00",
844
ApprovalPending=dbus.Boolean(False),
845
ApprovedByDefault=dbus.Boolean(False),
846
LastApprovalRequest="2019-01-03T00:00:00",
847
ApprovalDelay=30000,
848
ApprovalDuration=1000,
849
Checker=":",
850
ExtendedTimeout=900000,
851
Expires="2019-02-05T00:00:00",
852
LastCheckerStatus=-2)),
896
853
])
897
self.one_client = {self.client: self.client.attributes}
898
854
899
855
class TestPrintTableCmd(TestCmd):
900
856
def test_normal(self):
914
870
"""[1:-1]
915
871
self.assertEqual(output, expected_output)
916
872
def test_one_client(self):
917
output = PrintTableCmd().output(self.one_client)
873
output = PrintTableCmd().output({"foo": self.clients["foo"]})
918
874
expected_output = """
919
875
Name Enabled Timeout Last Successful Check
920
876
foo Yes 00:05:00 2019-02-03T00:00:00
978
934
json_data = json.loads(DumpJSONCmd().output(self.clients))
979
935
self.assertDictEqual(json_data, self.expected_json)
980
936
def test_one_client(self):
981
clients = self.one_client
937
clients = {"foo": self.clients["foo"]}
982
938
json_data = json.loads(DumpJSONCmd().output(clients))
983
939
expected_json = {"foo": self.expected_json["foo"]}
984
940
self.assertDictEqual(json_data, expected_json)
985
941
986
942
class TestIsEnabledCmd(TestCmd):
987
943
def test_is_enabled(self):
988
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
989
for client, properties in self.clients.items()))
944
self.assertTrue(all(IsEnabledCmd().is_enabled(client)
945
for client in self.clients.values()))
946
def test_is_enabled_does_get_attribute(self):
947
client = self.clients["foo"]
948
self.assertTrue(IsEnabledCmd().is_enabled(client))
949
self.assertListEqual(client.calls,
950
[("Get",
951
("se.recompile.Mandos.Client",
952
"Enabled",
953
"org.freedesktop.DBus.Properties"))])
990
954
def test_is_enabled_run_exits_successfully(self):
955
client = self.clients["foo"]
991
956
with self.assertRaises(SystemExit) as e:
992
IsEnabledCmd().run(None, self.one_client)
957
IsEnabledCmd().run(None, [client])
993
958
if e.exception.code is not None:
994
959
self.assertEqual(e.exception.code, 0)
995
960
else:
996
961
self.assertIsNone(e.exception.code)
997
962
def test_is_enabled_run_exits_with_failure(self):
998
self.client.attributes["Enabled"] = dbus.Boolean(False)
963
client = self.clients["foo"]
964
client["Enabled"] = dbus.Boolean(False)
999
965
with self.assertRaises(SystemExit) as e:
1000
IsEnabledCmd().run(None, self.one_client)
966
IsEnabledCmd().run(None, [client])
1001
967
if isinstance(e.exception.code, int):
1002
968
self.assertNotEqual(e.exception.code, 0)
1003
969
else:
1004
970
self.assertIsNotNone(e.exception.code)
1005
971
972
1006
973
class TestRemoveCmd(TestCmd):
1007
974
def test_remove(self):
975
client = self.clients["foo"]
1008
976
class MockMandos(object):
1009
977
def __init__(self):
1010
978
self.calls = []
1011
979
def RemoveClient(self, dbus_path):
1012
980
self.calls.append(("RemoveClient", (dbus_path,)))
1013
981
mandos = MockMandos()
1014
super(TestRemoveCmd, self).setUp()
1015
RemoveCmd().run(mandos, self.clients)
1016
self.assertEqual(len(mandos.calls), 2)
1017
for client in self.clients:
1018
self.assertIn(("RemoveClient",
1019
(client.__dbus_object_path__,)),
1020
mandos.calls)
1021
1022
class TestApproveCmd(TestCmd):
1023
def test_approve(self):
1024
ApproveCmd().run(None, self.clients)
1025
for client in self.clients:
1026
self.assertIn(("Approve", (True, client_interface)),
1027
client.calls)
1028
1029
class TestDenyCmd(TestCmd):
1030
def test_deny(self):
1031
DenyCmd().run(None, self.clients)
1032
for client in self.clients:
1033
self.assertIn(("Approve", (False, client_interface)),
1034
client.calls)
1035
1036
class TestEnableCmd(TestCmd):
1037
def test_enable(self):
1038
for client in self.clients:
1039
client.attributes["Enabled"] = False
1040
1041
EnableCmd().run(None, self.clients)
1042
1043
for client in self.clients:
1044
self.assertTrue(client.attributes["Enabled"])
1045
1046
class TestDisableCmd(TestCmd):
1047
def test_disable(self):
1048
DisableCmd().run(None, self.clients)
1049
1050
for client in self.clients:
1051
self.assertFalse(client.attributes["Enabled"])
1052
1053
class Unique(object):
1054
"""Class for objects which exist only to be unique objects, since
1055
unittest.mock.sentinel only exists in Python 3.3"""
1056
1057
class TestPropertyCmd(TestCmd):
1058
"""Abstract class for tests of PropertyCmd classes"""
1059
def runTest(self):
1060
if not hasattr(self, "command"):
1061
return
1062
values_to_get = getattr(self, "values_to_get",
1063
self.values_to_set)
1064
for value_to_set, value_to_get in zip(self.values_to_set,
1065
values_to_get):
1066
for client in self.clients:
1067
old_value = client.attributes[self.property]
1068
self.assertNotIsInstance(old_value, Unique)
1069
client.attributes[self.property] = Unique()
1070
self.run_command(value_to_set, self.clients)
1071
for client in self.clients:
1072
value = client.attributes[self.property]
1073
self.assertNotIsInstance(value, Unique)
1074
self.assertEqual(value, value_to_get)
1075
def run_command(self, value, clients):
1076
self.command().run(None, clients)
1077
1078
class TestBumpTimeoutCmd(TestPropertyCmd):
1079
command = BumpTimeoutCmd
1080
property = "LastCheckedOK"
1081
values_to_set = [""]
1082
1083
class TestStartCheckerCmd(TestPropertyCmd):
1084
command = StartCheckerCmd
1085
property = "CheckerRunning"
1086
values_to_set = [dbus.Boolean(True)]
1087
1088
class TestStopCheckerCmd(TestPropertyCmd):
1089
command = StopCheckerCmd
1090
property = "CheckerRunning"
1091
values_to_set = [dbus.Boolean(False)]
1092
1093
class TestApproveByDefaultCmd(TestPropertyCmd):
1094
command = ApproveByDefaultCmd
1095
property = "ApprovedByDefault"
1096
values_to_set = [dbus.Boolean(True)]
1097
1098
class TestDenyByDefaultCmd(TestPropertyCmd):
1099
command = DenyByDefaultCmd
1100
property = "ApprovedByDefault"
1101
values_to_set = [dbus.Boolean(False)]
1102
1103
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1104
"""Abstract class for tests of PropertyCmd classes using the
1105
ValueArgumentMixIn"""
1106
def runTest(self):
1107
if type(self) is TestValueArgumentPropertyCmd:
1108
return
1109
return super(TestValueArgumentPropertyCmd, self).runTest()
1110
def run_command(self, value, clients):
1111
self.command(value).run(None, clients)
1112
1113
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1114
command = SetCheckerCmd
1115
property = "Checker"
1116
values_to_set = ["", ":", "fping -q -- %s"]
1117
1118
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1119
command = SetHostCmd
1120
property = "Host"
1121
values_to_set = ["192.0.2.3", "foo.example.org"]
1122
1123
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1124
command = SetSecretCmd
1125
property = "Secret"
1126
values_to_set = [io.BytesIO(b""),
1127
io.BytesIO(b"secret\0xyzzy\nbar")]
1128
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1129
1130
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1131
command = SetTimeoutCmd
1132
property = "Timeout"
1133
values_to_set = [datetime.timedelta(),
1134
datetime.timedelta(minutes=5),
1135
datetime.timedelta(seconds=1),
1136
datetime.timedelta(weeks=1),
1137
datetime.timedelta(weeks=52)]
1138
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1139
1140
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1141
command = SetExtendedTimeoutCmd
1142
property = "ExtendedTimeout"
1143
values_to_set = [datetime.timedelta(),
1144
datetime.timedelta(minutes=5),
1145
datetime.timedelta(seconds=1),
1146
datetime.timedelta(weeks=1),
1147
datetime.timedelta(weeks=52)]
1148
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1149
1150
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1151
command = SetIntervalCmd
1152
property = "Interval"
1153
values_to_set = [datetime.timedelta(),
1154
datetime.timedelta(minutes=5),
1155
datetime.timedelta(seconds=1),
1156
datetime.timedelta(weeks=1),
1157
datetime.timedelta(weeks=52)]
1158
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1159
1160
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1161
command = SetApprovalDelayCmd
1162
property = "ApprovalDelay"
1163
values_to_set = [datetime.timedelta(),
1164
datetime.timedelta(minutes=5),
1165
datetime.timedelta(seconds=1),
1166
datetime.timedelta(weeks=1),
1167
datetime.timedelta(weeks=52)]
1168
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1169
1170
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1171
command = SetApprovalDurationCmd
1172
property = "ApprovalDuration"
1173
values_to_set = [datetime.timedelta(),
1174
datetime.timedelta(minutes=5),
1175
datetime.timedelta(seconds=1),
1176
datetime.timedelta(weeks=1),
1177
datetime.timedelta(weeks=52)]
1178
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1179
1180
class Test_command_from_options(unittest.TestCase):
1181
def setUp(self):
1182
self.parser = argparse.ArgumentParser()
1183
add_command_line_options(self.parser)
1184
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1185
"""Assert that parsing ARGS should result in an instance of
1186
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1187
options = self.parser.parse_args(args)
1188
check_option_syntax(self.parser, options)
1189
commands = commands_from_options(options)
1190
self.assertEqual(len(commands), 1)
1191
command = commands[0]
1192
self.assertIsInstance(command, command_cls)
1193
for key, value in cmd_attrs.items():
1194
self.assertEqual(getattr(command, key), value)
1195
def test_print_table(self):
1196
self.assert_command_from_args([], PrintTableCmd,
1197
verbose=False)
1198
1199
def test_print_table_verbose(self):
1200
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1201
verbose=True)
1202
1203
def test_print_table_verbose_short(self):
1204
self.assert_command_from_args(["-v"], PrintTableCmd,
1205
verbose=True)
1206
1207
def test_enable(self):
1208
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1209
1210
def test_enable_short(self):
1211
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1212
1213
def test_disable(self):
1214
self.assert_command_from_args(["--disable", "foo"],
1215
DisableCmd)
1216
1217
def test_disable_short(self):
1218
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1219
1220
def test_bump_timeout(self):
1221
self.assert_command_from_args(["--bump-timeout", "foo"],
1222
BumpTimeoutCmd)
1223
1224
def test_bump_timeout_short(self):
1225
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1226
1227
def test_start_checker(self):
1228
self.assert_command_from_args(["--start-checker", "foo"],
1229
StartCheckerCmd)
1230
1231
def test_stop_checker(self):
1232
self.assert_command_from_args(["--stop-checker", "foo"],
1233
StopCheckerCmd)
1234
1235
def test_remove(self):
1236
self.assert_command_from_args(["--remove", "foo"],
1237
RemoveCmd)
1238
1239
def test_remove_short(self):
1240
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1241
1242
def test_checker(self):
1243
self.assert_command_from_args(["--checker", ":", "foo"],
1244
SetCheckerCmd, value_to_set=":")
1245
1246
def test_checker_empty(self):
1247
self.assert_command_from_args(["--checker", "", "foo"],
1248
SetCheckerCmd, value_to_set="")
1249
1250
def test_checker_short(self):
1251
self.assert_command_from_args(["-c", ":", "foo"],
1252
SetCheckerCmd, value_to_set=":")
1253
1254
def test_timeout(self):
1255
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1256
SetTimeoutCmd,
1257
value_to_set=300000)
1258
1259
def test_timeout_short(self):
1260
self.assert_command_from_args(["-t", "PT5M", "foo"],
1261
SetTimeoutCmd,
1262
value_to_set=300000)
1263
1264
def test_extended_timeout(self):
1265
self.assert_command_from_args(["--extended-timeout", "PT15M",
1266
"foo"],
1267
SetExtendedTimeoutCmd,
1268
value_to_set=900000)
1269
1270
def test_interval(self):
1271
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1272
SetIntervalCmd,
1273
value_to_set=120000)
1274
1275
def test_interval_short(self):
1276
self.assert_command_from_args(["-i", "PT2M", "foo"],
1277
SetIntervalCmd,
1278
value_to_set=120000)
1279
1280
def test_approve_by_default(self):
1281
self.assert_command_from_args(["--approve-by-default", "foo"],
1282
ApproveByDefaultCmd)
1283
1284
def test_deny_by_default(self):
1285
self.assert_command_from_args(["--deny-by-default", "foo"],
1286
DenyByDefaultCmd)
1287
1288
def test_approval_delay(self):
1289
self.assert_command_from_args(["--approval-delay", "PT30S",
1290
"foo"], SetApprovalDelayCmd,
1291
value_to_set=30000)
1292
1293
def test_approval_duration(self):
1294
self.assert_command_from_args(["--approval-duration", "PT1S",
1295
"foo"], SetApprovalDurationCmd,
1296
value_to_set=1000)
1297
1298
def test_host(self):
1299
self.assert_command_from_args(["--host", "foo.example.org",
1300
"foo"], SetHostCmd,
1301
value_to_set="foo.example.org")
1302
1303
def test_host_short(self):
1304
self.assert_command_from_args(["-H", "foo.example.org",
1305
"foo"], SetHostCmd,
1306
value_to_set="foo.example.org")
1307
1308
def test_secret_devnull(self):
1309
self.assert_command_from_args(["--secret", os.path.devnull,
1310
"foo"], SetSecretCmd,
1311
value_to_set=b"")
1312
1313
def test_secret_tempfile(self):
1314
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1315
value = b"secret\0xyzzy\nbar"
1316
f.write(value)
1317
f.seek(0)
1318
self.assert_command_from_args(["--secret", f.name,
1319
"foo"], SetSecretCmd,
1320
value_to_set=value)
1321
1322
def test_secret_devnull_short(self):
1323
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1324
SetSecretCmd, value_to_set=b"")
1325
1326
def test_secret_tempfile_short(self):
1327
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1328
value = b"secret\0xyzzy\nbar"
1329
f.write(value)
1330
f.seek(0)
1331
self.assert_command_from_args(["-s", f.name, "foo"],
1332
SetSecretCmd,
1333
value_to_set=value)
1334
1335
def test_approve(self):
1336
self.assert_command_from_args(["--approve", "foo"],
1337
ApproveCmd)
1338
1339
def test_approve_short(self):
1340
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1341
1342
def test_deny(self):
1343
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1344
1345
def test_deny_short(self):
1346
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1347
1348
def test_dump_json(self):
1349
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1350
1351
def test_is_enabled(self):
1352
self.assert_command_from_args(["--is-enabled", "foo"],
1353
IsEnabledCmd)
1354
1355
def test_is_enabled_short(self):
1356
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1357
1358
1359
class Test_check_option_syntax(unittest.TestCase):
1360
# This mostly corresponds to the definition from has_actions() in
1361
# check_option_syntax()
1362
actions = {
1363
# The actual values set here are not that important, but we do
1364
# at least stick to the correct types, even though they are
1365
# never used
1366
"enable": True,
1367
"disable": True,
1368
"bump_timeout": True,
1369
"start_checker": True,
1370
"stop_checker": True,
1371
"is_enabled": True,
1372
"remove": True,
1373
"checker": "x",
1374
"timeout": datetime.timedelta(),
1375
"extended_timeout": datetime.timedelta(),
1376
"interval": datetime.timedelta(),
1377
"approved_by_default": True,
1378
"approval_delay": datetime.timedelta(),
1379
"approval_duration": datetime.timedelta(),
1380
"host": "x",
1381
"secret": io.BytesIO(b"x"),
1382
"approve": True,
1383
"deny": True,
1384
}
1385
1386
def setUp(self):
1387
self.parser = argparse.ArgumentParser()
1388
add_command_line_options(self.parser)
1389
1390
@contextlib.contextmanager
1391
def assertParseError(self):
1392
with self.assertRaises(SystemExit) as e:
1393
with self.temporarily_suppress_stderr():
1394
yield
1395
# Exit code from argparse is guaranteed to be "2". Reference:
1396
# https://docs.python.org/3/library/argparse.html#exiting-methods
1397
self.assertEqual(e.exception.code, 2)
1398
1399
@staticmethod
1400
@contextlib.contextmanager
1401
def temporarily_suppress_stderr():
1402
null = os.open(os.path.devnull, os.O_RDWR)
1403
stderrcopy = os.dup(sys.stderr.fileno())
1404
os.dup2(null, sys.stderr.fileno())
1405
os.close(null)
1406
try:
1407
yield
1408
finally:
1409
# restore stderr
1410
os.dup2(stderrcopy, sys.stderr.fileno())
1411
os.close(stderrcopy)
1412
1413
def check_option_syntax(self, options):
1414
check_option_syntax(self.parser, options)
1415
1416
def test_actions_requires_client_or_all(self):
1417
for action, value in self.actions.items():
1418
options = self.parser.parse_args()
1419
setattr(options, action, value)
1420
with self.assertParseError():
1421
self.check_option_syntax(options)
1422
1423
def test_actions_conflicts_with_verbose(self):
1424
for action, value in self.actions.items():
1425
options = self.parser.parse_args()
1426
setattr(options, action, value)
1427
options.verbose = True
1428
with self.assertParseError():
1429
self.check_option_syntax(options)
1430
1431
def test_dump_json_conflicts_with_verbose(self):
1432
options = self.parser.parse_args()
1433
options.dump_json = True
1434
options.verbose = True
1435
with self.assertParseError():
1436
self.check_option_syntax(options)
1437
1438
def test_dump_json_conflicts_with_action(self):
1439
for action, value in self.actions.items():
1440
options = self.parser.parse_args()
1441
setattr(options, action, value)
1442
options.dump_json = True
1443
with self.assertParseError():
1444
self.check_option_syntax(options)
1445
1446
def test_all_can_not_be_alone(self):
1447
options = self.parser.parse_args()
1448
options.all = True
1449
with self.assertParseError():
1450
self.check_option_syntax(options)
1451
1452
def test_all_is_ok_with_any_action(self):
1453
for action, value in self.actions.items():
1454
options = self.parser.parse_args()
1455
setattr(options, action, value)
1456
options.all = True
1457
self.check_option_syntax(options)
1458
1459
def test_is_enabled_fails_without_client(self):
1460
options = self.parser.parse_args()
1461
options.is_enabled = True
1462
with self.assertParseError():
1463
self.check_option_syntax(options)
1464
1465
def test_is_enabled_works_with_one_client(self):
1466
options = self.parser.parse_args()
1467
options.is_enabled = True
1468
options.client = ["foo"]
1469
self.check_option_syntax(options)
1470
1471
def test_is_enabled_fails_with_two_clients(self):
1472
options = self.parser.parse_args()
1473
options.is_enabled = True
1474
options.client = ["foo", "barbar"]
1475
with self.assertParseError():
1476
self.check_option_syntax(options)
982
RemoveCmd().run(mandos, [client])
983
self.assertEqual(len(mandos.calls), 1)
984
self.assertListEqual(mandos.calls,
985
[("RemoveClient",
986
(client.__dbus_object_path__,))])
1477
987
1478
988
1479
989
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0