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 1021
- compare with another revision
- download diff
- download tarball
- view history from revision 1021
- Committer: Teddy Hogeborn
- Date: 2019-03-04 20:50:40 UTC
- Revision ID: teddy@recompile.se-20190304205040-c0ecwnigjedch1j9
mandos-ctl: Add test for EnableCmd and DisableCmd
* mandos-ctl (TestEnableCmd, TestDisableCmd): New.
* mandos-ctl (TestEnableCmd, TestDisableCmd): 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
45
48
46
import dbus
49
47
316
314
@value_to_set.setter
317
315
def value_to_set(self, value):
318
316
"""When setting, convert value to a datetime.timedelta"""
319
self._vts = int(round(value.total_seconds() * 1000))
317
self._vts = string_to_delta(value).total_seconds() * 1000
320
318
321
319
# Actual (non-abstract) command classes
322
320
325
323
self.verbose = verbose
326
324
327
325
def output(self, clients):
328
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
329
keywords = default_keywords
330
326
if self.verbose:
331
327
keywords = self.all_keywords
328
else:
329
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
332
330
return str(self.TableOfClients(clients.values(), keywords))
333
331
334
332
class TableOfClients(object):
477
475
property = "Host"
478
476
479
477
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
480
@property
481
def value_to_set(self):
482
return self._vts
483
@value_to_set.setter
484
def value_to_set(self, value):
485
"""When setting, read data from supplied file object"""
486
self._vts = value.read()
487
value.close()
488
478
property = "Secret"
489
479
490
480
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
505
495
MillisecondsValueArgumentMixIn):
506
496
property = "ApprovalDuration"
507
497
498
def has_actions(options):
499
return any((options.enable,
500
options.disable,
501
options.bump_timeout,
502
options.start_checker,
503
options.stop_checker,
504
options.is_enabled,
505
options.remove,
506
options.checker is not None,
507
options.timeout is not None,
508
options.extended_timeout is not None,
509
options.interval is not None,
510
options.approved_by_default is not None,
511
options.approval_delay is not None,
512
options.approval_duration is not None,
513
options.host is not None,
514
options.secret is not None,
515
options.approve,
516
options.deny))
517
508
518
def add_command_line_options(parser):
509
519
parser.add_argument("--version", action="version",
510
520
version="%(prog)s {}".format(version),
536
546
help="Remove client")
537
547
parser.add_argument("-c", "--checker",
538
548
help="Set checker command for client")
539
parser.add_argument("-t", "--timeout", type=string_to_delta,
549
parser.add_argument("-t", "--timeout",
540
550
help="Set timeout for client")
541
parser.add_argument("--extended-timeout", type=string_to_delta,
551
parser.add_argument("--extended-timeout",
542
552
help="Set extended timeout for client")
543
parser.add_argument("-i", "--interval", type=string_to_delta,
553
parser.add_argument("-i", "--interval",
544
554
help="Set checker interval for client")
545
555
approve_deny_default = parser.add_mutually_exclusive_group()
546
556
approve_deny_default.add_argument(
551
561
"--deny-by-default", action="store_false",
552
562
dest="approved_by_default",
553
563
help="Set client to be denied by default")
554
parser.add_argument("--approval-delay", type=string_to_delta,
564
parser.add_argument("--approval-delay",
555
565
help="Set delay before client approve/deny")
556
parser.add_argument("--approval-duration", type=string_to_delta,
566
parser.add_argument("--approval-duration",
557
567
help="Set duration of one client approval")
558
568
parser.add_argument("-H", "--host", help="Set host for client")
559
569
parser.add_argument("-s", "--secret",
570
580
parser.add_argument("client", nargs="*", help="Client name")
571
581
572
582
573
def commands_from_options(options):
583
def commands_and_clients_from_options(options):
574
584
575
585
commands = []
576
586
584
594
commands.append(DisableCmd())
585
595
586
596
if options.bump_timeout:
587
commands.append(BumpTimeoutCmd())
597
commands.append(BumpTimeoutCmd(options.bump_timeout))
588
598
589
599
if options.start_checker:
590
600
commands.append(StartCheckerCmd())
599
609
commands.append(RemoveCmd())
600
610
601
611
if options.checker is not None:
602
commands.append(SetCheckerCmd(options.checker))
612
commands.append(SetCheckerCmd())
603
613
604
614
if options.timeout is not None:
605
615
commands.append(SetTimeoutCmd(options.timeout))
609
619
SetExtendedTimeoutCmd(options.extended_timeout))
610
620
611
621
if options.interval is not None:
612
commands.append(SetIntervalCmd(options.interval))
622
command.append(SetIntervalCmd(options.interval))
613
623
614
624
if options.approved_by_default is not None:
615
625
if options.approved_by_default:
616
commands.append(ApproveByDefaultCmd())
626
command.append(ApproveByDefaultCmd())
617
627
else:
618
commands.append(DenyByDefaultCmd())
628
command.append(DenyByDefaultCmd())
619
629
620
630
if options.approval_delay is not None:
621
commands.append(SetApprovalDelayCmd(options.approval_delay))
631
command.append(SetApprovalDelayCmd(options.approval_delay))
622
632
623
633
if options.approval_duration is not None:
624
commands.append(
634
command.append(
625
635
SetApprovalDurationCmd(options.approval_duration))
626
636
627
637
if options.host is not None:
628
commands.append(SetHostCmd(options.host))
638
command.append(SetHostCmd(options.host))
629
639
630
640
if options.secret is not None:
631
commands.append(SetSecretCmd(options.secret))
641
command.append(SetSecretCmd(options.secret))
632
642
633
643
if options.approve:
634
644
commands.append(ApproveCmd())
641
651
if not commands:
642
652
commands.append(PrintTableCmd(verbose=options.verbose))
643
653
644
return commands
645
646
647
def check_option_syntax(parser, options):
648
649
def has_actions(options):
650
return any((options.enable,
651
options.disable,
652
options.bump_timeout,
653
options.start_checker,
654
options.stop_checker,
655
options.is_enabled,
656
options.remove,
657
options.checker is not None,
658
options.timeout is not None,
659
options.extended_timeout is not None,
660
options.interval is not None,
661
options.approved_by_default is not None,
662
options.approval_delay is not None,
663
options.approval_duration is not None,
664
options.host is not None,
665
options.secret is not None,
666
options.approve,
667
options.deny))
654
return commands, options.client
655
656
657
def main():
658
parser = argparse.ArgumentParser()
659
660
add_command_line_options(parser)
661
662
options = parser.parse_args()
668
663
669
664
if has_actions(options) and not (options.client or options.all):
670
665
parser.error("Options require clients names or --all.")
678
673
if options.is_enabled and len(options.client) > 1:
679
674
parser.error("--is-enabled requires exactly one client")
680
675
681
682
def main():
683
parser = argparse.ArgumentParser()
684
685
add_command_line_options(parser)
686
687
options = parser.parse_args()
688
689
check_option_syntax(parser, options)
690
691
clientnames = options.client
676
commands, clientnames = commands_and_clients_from_options(options)
692
677
693
678
try:
694
679
bus = dbus.SystemBus()
741
726
sys.exit(1)
742
727
743
728
# Run all commands on clients
744
commands = commands_from_options(options)
745
729
for command in commands:
746
730
command.run(mandos_serv, clients)
747
731
761
745
762
746
class Test_string_to_delta(unittest.TestCase):
763
747
def test_handles_basic_rfc3339(self):
764
self.assertEqual(string_to_delta("PT0S"),
765
datetime.timedelta())
766
self.assertEqual(string_to_delta("P0D"),
767
datetime.timedelta())
768
self.assertEqual(string_to_delta("PT1S"),
769
datetime.timedelta(0, 1))
770
748
self.assertEqual(string_to_delta("PT2H"),
771
749
datetime.timedelta(0, 7200))
772
750
def test_falls_back_to_pre_1_6_1_with_warning(self):
807
785
testcase.assertEqual(dbus_interface,
808
786
dbus.PROPERTIES_IFACE)
809
787
self.attributes[property] = value
788
self.calls.append(("Set", (interface, property, value,
789
dbus_interface)))
810
790
def Get(self, interface, property, dbus_interface):
811
791
testcase.assertEqual(interface, client_interface)
812
792
testcase.assertEqual(dbus_interface,
813
793
dbus.PROPERTIES_IFACE)
794
self.calls.append(("Get", (interface, property,
795
dbus_interface)))
814
796
return self.attributes[property]
815
797
def Approve(self, approve, dbus_interface):
816
798
testcase.assertEqual(dbus_interface, client_interface)
1025
1007
for client in self.clients:
1026
1008
self.assertFalse(client.attributes["Enabled"])
1027
1009
1028
class Unique(object):
1029
"""Class for objects which exist only to be unique objects, since
1030
unittest.mock.sentinel only exists in Python 3.3"""
1031
1032
class TestPropertyCmd(TestCmd):
1033
"""Abstract class for tests of PropertyCmd classes"""
1034
def runTest(self):
1035
if not hasattr(self, "command"):
1036
return
1037
values_to_get = getattr(self, "values_to_get",
1038
self.values_to_set)
1039
for value_to_set, value_to_get in zip(self.values_to_set,
1040
values_to_get):
1041
for client in self.clients:
1042
old_value = client.attributes[self.property]
1043
self.assertNotIsInstance(old_value, Unique)
1044
client.attributes[self.property] = Unique()
1045
self.run_command(value_to_set, self.clients)
1046
for client in self.clients:
1047
value = client.attributes[self.property]
1048
self.assertNotIsInstance(value, Unique)
1049
self.assertEqual(value, value_to_get)
1050
def run_command(self, value, clients):
1051
self.command().run(None, clients)
1052
1053
class TestBumpTimeoutCmd(TestPropertyCmd):
1054
command = BumpTimeoutCmd
1055
property = "LastCheckedOK"
1056
values_to_set = [""]
1057
1058
class TestStartCheckerCmd(TestPropertyCmd):
1059
command = StartCheckerCmd
1060
property = "CheckerRunning"
1061
values_to_set = [dbus.Boolean(True)]
1062
1063
class TestStopCheckerCmd(TestPropertyCmd):
1064
command = StopCheckerCmd
1065
property = "CheckerRunning"
1066
values_to_set = [dbus.Boolean(False)]
1067
1068
class TestApproveByDefaultCmd(TestPropertyCmd):
1069
command = ApproveByDefaultCmd
1070
property = "ApprovedByDefault"
1071
values_to_set = [dbus.Boolean(True)]
1072
1073
class TestDenyByDefaultCmd(TestPropertyCmd):
1074
command = DenyByDefaultCmd
1075
property = "ApprovedByDefault"
1076
values_to_set = [dbus.Boolean(False)]
1077
1078
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1079
"""Abstract class for tests of PropertyCmd classes using the
1080
ValueArgumentMixIn"""
1081
def runTest(self):
1082
if type(self) is TestValueArgumentPropertyCmd:
1083
return
1084
return super(TestValueArgumentPropertyCmd, self).runTest()
1085
def run_command(self, value, clients):
1086
self.command(value).run(None, clients)
1087
1088
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1089
command = SetCheckerCmd
1090
property = "Checker"
1091
values_to_set = ["", ":", "fping -q -- %s"]
1092
1093
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1094
command = SetHostCmd
1095
property = "Host"
1096
values_to_set = ["192.0.2.3", "foo.example.org"]
1097
1098
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1099
command = SetSecretCmd
1100
property = "Secret"
1101
values_to_set = [open("/dev/null", "rb"),
1102
io.BytesIO(b"secret\0xyzzy\nbar")]
1103
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1104
1105
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1106
command = SetTimeoutCmd
1107
property = "Timeout"
1108
values_to_set = [datetime.timedelta(),
1109
datetime.timedelta(minutes=5),
1110
datetime.timedelta(seconds=1),
1111
datetime.timedelta(weeks=1),
1112
datetime.timedelta(weeks=52)]
1113
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1114
1115
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1116
command = SetExtendedTimeoutCmd
1117
property = "ExtendedTimeout"
1118
values_to_set = [datetime.timedelta(),
1119
datetime.timedelta(minutes=5),
1120
datetime.timedelta(seconds=1),
1121
datetime.timedelta(weeks=1),
1122
datetime.timedelta(weeks=52)]
1123
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1124
1125
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1126
command = SetIntervalCmd
1127
property = "Interval"
1128
values_to_set = [datetime.timedelta(),
1129
datetime.timedelta(minutes=5),
1130
datetime.timedelta(seconds=1),
1131
datetime.timedelta(weeks=1),
1132
datetime.timedelta(weeks=52)]
1133
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1134
1135
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1136
command = SetApprovalDelayCmd
1137
property = "ApprovalDelay"
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]
1144
1145
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1146
command = SetApprovalDurationCmd
1147
property = "ApprovalDuration"
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]
1154
1155
class Test_command_from_options(unittest.TestCase):
1156
def setUp(self):
1157
self.parser = argparse.ArgumentParser()
1158
add_command_line_options(self.parser)
1159
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1160
"""Assert that parsing ARGS should result in an instance of
1161
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1162
options = self.parser.parse_args(args)
1163
check_option_syntax(self.parser, options)
1164
commands = commands_from_options(options)
1165
self.assertEqual(len(commands), 1)
1166
command = commands[0]
1167
self.assertIsInstance(command, command_cls)
1168
for key, value in cmd_attrs.items():
1169
self.assertEqual(getattr(command, key), value)
1170
def test_print_table(self):
1171
self.assert_command_from_args([], PrintTableCmd,
1172
verbose=False)
1173
1174
def test_print_table_verbose(self):
1175
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1176
verbose=True)
1177
1178
def test_print_table_verbose_short(self):
1179
self.assert_command_from_args(["-v"], PrintTableCmd,
1180
verbose=True)
1181
1182
def test_enable(self):
1183
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1184
1185
def test_enable_short(self):
1186
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1187
1188
def test_disable(self):
1189
self.assert_command_from_args(["--disable", "foo"],
1190
DisableCmd)
1191
1192
def test_disable_short(self):
1193
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1194
1195
def test_bump_timeout(self):
1196
self.assert_command_from_args(["--bump-timeout", "foo"],
1197
BumpTimeoutCmd)
1198
1199
def test_bump_timeout_short(self):
1200
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1201
1202
def test_start_checker(self):
1203
self.assert_command_from_args(["--start-checker", "foo"],
1204
StartCheckerCmd)
1205
1206
def test_stop_checker(self):
1207
self.assert_command_from_args(["--stop-checker", "foo"],
1208
StopCheckerCmd)
1209
1210
def test_remove(self):
1211
self.assert_command_from_args(["--remove", "foo"],
1212
RemoveCmd)
1213
1214
def test_remove_short(self):
1215
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1216
1217
def test_checker(self):
1218
self.assert_command_from_args(["--checker", ":", "foo"],
1219
SetCheckerCmd, value_to_set=":")
1220
1221
def test_checker_empty(self):
1222
self.assert_command_from_args(["--checker", "", "foo"],
1223
SetCheckerCmd, value_to_set="")
1224
1225
def test_checker_short(self):
1226
self.assert_command_from_args(["-c", ":", "foo"],
1227
SetCheckerCmd, value_to_set=":")
1228
1229
def test_timeout(self):
1230
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1231
SetTimeoutCmd,
1232
value_to_set=300000)
1233
1234
def test_timeout_short(self):
1235
self.assert_command_from_args(["-t", "PT5M", "foo"],
1236
SetTimeoutCmd,
1237
value_to_set=300000)
1238
1239
def test_extended_timeout(self):
1240
self.assert_command_from_args(["--extended-timeout", "PT15M",
1241
"foo"],
1242
SetExtendedTimeoutCmd,
1243
value_to_set=900000)
1244
1245
def test_interval(self):
1246
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1247
SetIntervalCmd,
1248
value_to_set=120000)
1249
1250
def test_interval_short(self):
1251
self.assert_command_from_args(["-i", "PT2M", "foo"],
1252
SetIntervalCmd,
1253
value_to_set=120000)
1254
1255
def test_approve_by_default(self):
1256
self.assert_command_from_args(["--approve-by-default", "foo"],
1257
ApproveByDefaultCmd)
1258
1259
def test_deny_by_default(self):
1260
self.assert_command_from_args(["--deny-by-default", "foo"],
1261
DenyByDefaultCmd)
1262
1263
def test_approval_delay(self):
1264
self.assert_command_from_args(["--approval-delay", "PT30S",
1265
"foo"], SetApprovalDelayCmd,
1266
value_to_set=30000)
1267
1268
def test_approval_duration(self):
1269
self.assert_command_from_args(["--approval-duration", "PT1S",
1270
"foo"], SetApprovalDurationCmd,
1271
value_to_set=1000)
1272
1273
def test_host(self):
1274
self.assert_command_from_args(["--host", "foo.example.org",
1275
"foo"], SetHostCmd,
1276
value_to_set="foo.example.org")
1277
1278
def test_host_short(self):
1279
self.assert_command_from_args(["-H", "foo.example.org",
1280
"foo"], SetHostCmd,
1281
value_to_set="foo.example.org")
1282
1283
def test_secret_devnull(self):
1284
self.assert_command_from_args(["--secret", os.path.devnull,
1285
"foo"], SetSecretCmd,
1286
value_to_set=b"")
1287
1288
def test_secret_tempfile(self):
1289
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1290
value = b"secret\0xyzzy\nbar"
1291
f.write(value)
1292
f.seek(0)
1293
self.assert_command_from_args(["--secret", f.name,
1294
"foo"], SetSecretCmd,
1295
value_to_set=value)
1296
1297
def test_secret_devnull_short(self):
1298
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1299
SetSecretCmd, value_to_set=b"")
1300
1301
def test_secret_tempfile_short(self):
1302
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1303
value = b"secret\0xyzzy\nbar"
1304
f.write(value)
1305
f.seek(0)
1306
self.assert_command_from_args(["-s", f.name, "foo"],
1307
SetSecretCmd,
1308
value_to_set=value)
1309
1310
def test_approve(self):
1311
self.assert_command_from_args(["--approve", "foo"],
1312
ApproveCmd)
1313
1314
def test_approve_short(self):
1315
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1316
1317
def test_deny(self):
1318
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1319
1320
def test_deny_short(self):
1321
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1322
1323
def test_dump_json(self):
1324
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1325
1326
def test_is_enabled(self):
1327
self.assert_command_from_args(["--is-enabled", "foo"],
1328
IsEnabledCmd)
1329
1330
def test_is_enabled_short(self):
1331
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1332
1333
1010
1334
1011
1335
1012
def should_only_run_tests():
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0