229
268
value += datetime.timedelta(0, 0, 0, int(num))
232
def print_clients(clients, keywords):
233
def valuetostring(value, keyword):
234
if type(value) is dbus.Boolean:
235
return "Yes" if value else "No"
236
if keyword in ("Timeout", "Interval", "ApprovalDelay",
237
"ApprovalDuration", "ExtendedTimeout"):
238
return milliseconds_to_string(value)
241
# Create format string to print table rows
242
format_string = " ".join("{{{key}:{width}}}".format(
243
width = max(len(tablewords[key]),
244
max(len(valuetostring(client[key],
248
key = key) for key in keywords)
250
print(format_string.format(**tablewords))
251
for client in clients:
252
print(format_string.format(**dict((key,
253
valuetostring(client[key],
255
for key in keywords)))
272
## Classes for commands.
274
# Abstract classes first
275
class Command(object):
276
"""Abstract class for commands"""
277
def run(self, mandos, clients):
278
"""Normal commands should implement run_on_one_client(), but
279
commands which want to operate on all clients at the same time
280
can override this run() method instead."""
282
for client, properties in clients.items():
283
self.run_on_one_client(client, properties)
285
class PrintCmd(Command):
286
"""Abstract class for commands printing client details"""
287
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
288
"Created", "Interval", "Host", "KeyID",
289
"Fingerprint", "CheckerRunning", "LastEnabled",
290
"ApprovalPending", "ApprovedByDefault",
291
"LastApprovalRequest", "ApprovalDelay",
292
"ApprovalDuration", "Checker", "ExtendedTimeout",
293
"Expires", "LastCheckerStatus")
294
def run(self, mandos, clients):
295
print(self.output(clients))
297
class PropertyCmd(Command):
298
"""Abstract class for Actions for setting one client property"""
299
def run_on_one_client(self, client, properties):
300
"""Set the Client's D-Bus property"""
301
client.Set(client_interface, self.property, self.value_to_set,
302
dbus_interface=dbus.PROPERTIES_IFACE)
304
class ValueArgumentMixIn(object):
305
"""Mixin class for commands taking a value as argument"""
306
def __init__(self, value):
307
self.value_to_set = value
309
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
310
"""Mixin class for commands taking a value argument as
313
def value_to_set(self):
316
def value_to_set(self, value):
317
"""When setting, convert value to a datetime.timedelta"""
318
self._vts = string_to_delta(value).total_seconds() * 1000
320
# Actual (non-abstract) command classes
322
class PrintTableCmd(PrintCmd):
323
def __init__(self, verbose=False):
324
self.verbose = verbose
326
def output(self, clients):
327
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
328
keywords = default_keywords
330
keywords = self.all_keywords
331
return str(self.TableOfClients(clients.values(), keywords))
333
class TableOfClients(object):
336
"Enabled": "Enabled",
337
"Timeout": "Timeout",
338
"LastCheckedOK": "Last Successful Check",
339
"LastApprovalRequest": "Last Approval Request",
340
"Created": "Created",
341
"Interval": "Interval",
343
"Fingerprint": "Fingerprint",
345
"CheckerRunning": "Check Is Running",
346
"LastEnabled": "Last Enabled",
347
"ApprovalPending": "Approval Is Pending",
348
"ApprovedByDefault": "Approved By Default",
349
"ApprovalDelay": "Approval Delay",
350
"ApprovalDuration": "Approval Duration",
351
"Checker": "Checker",
352
"ExtendedTimeout": "Extended Timeout",
353
"Expires": "Expires",
354
"LastCheckerStatus": "Last Checker Status",
357
def __init__(self, clients, keywords, tableheaders=None):
358
self.clients = clients
359
self.keywords = keywords
360
if tableheaders is not None:
361
self.tableheaders = tableheaders
364
return "\n".join(self.rows())
366
if sys.version_info.major == 2:
367
__unicode__ = __str__
369
return str(self).encode(locale.getpreferredencoding())
372
format_string = self.row_formatting_string()
373
rows = [self.header_line(format_string)]
374
rows.extend(self.client_line(client, format_string)
375
for client in self.clients)
378
def row_formatting_string(self):
379
"Format string used to format table rows"
380
return " ".join("{{{key}:{width}}}".format(
381
width=max(len(self.tableheaders[key]),
382
*(len(self.string_from_client(client, key))
383
for client in self.clients)),
385
for key in self.keywords)
387
def string_from_client(self, client, key):
388
return self.valuetostring(client[key], key)
391
def valuetostring(value, keyword):
392
if isinstance(value, dbus.Boolean):
393
return "Yes" if value else "No"
394
if keyword in ("Timeout", "Interval", "ApprovalDelay",
395
"ApprovalDuration", "ExtendedTimeout"):
396
return milliseconds_to_string(value)
399
def header_line(self, format_string):
400
return format_string.format(**self.tableheaders)
402
def client_line(self, client, format_string):
403
return format_string.format(
404
**{key: self.string_from_client(client, key)
405
for key in self.keywords})
409
class DumpJSONCmd(PrintCmd):
410
def output(self, clients):
411
data = {client["Name"]:
412
{key: self.dbus_boolean_to_bool(client[key])
413
for key in self.all_keywords}
414
for client in clients.values()}
415
return json.dumps(data, indent=4, separators=(',', ': '))
417
def dbus_boolean_to_bool(value):
418
if isinstance(value, dbus.Boolean):
422
class IsEnabledCmd(Command):
423
def run_on_one_client(self, client, properties):
424
if self.is_enabled(client, properties):
427
def is_enabled(self, client, properties):
428
return bool(properties["Enabled"])
430
class RemoveCmd(Command):
431
def run_on_one_client(self, client, properties):
432
self.mandos.RemoveClient(client.__dbus_object_path__)
434
class ApproveCmd(Command):
435
def run_on_one_client(self, client, properties):
436
client.Approve(dbus.Boolean(True),
437
dbus_interface=client_interface)
439
class DenyCmd(Command):
440
def run_on_one_client(self, client, properties):
441
client.Approve(dbus.Boolean(False),
442
dbus_interface=client_interface)
444
class EnableCmd(PropertyCmd):
446
value_to_set = dbus.Boolean(True)
448
class DisableCmd(PropertyCmd):
450
value_to_set = dbus.Boolean(False)
452
class BumpTimeoutCmd(PropertyCmd):
453
property = "LastCheckedOK"
456
class StartCheckerCmd(PropertyCmd):
457
property = "CheckerRunning"
458
value_to_set = dbus.Boolean(True)
460
class StopCheckerCmd(PropertyCmd):
461
property = "CheckerRunning"
462
value_to_set = dbus.Boolean(False)
464
class ApproveByDefaultCmd(PropertyCmd):
465
property = "ApprovedByDefault"
466
value_to_set = dbus.Boolean(True)
468
class DenyByDefaultCmd(PropertyCmd):
469
property = "ApprovedByDefault"
470
value_to_set = dbus.Boolean(False)
472
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
475
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
478
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
480
def value_to_set(self):
483
def value_to_set(self, value):
484
"""When setting, read data from supplied file object"""
485
self._vts = value.read()
489
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
492
class SetExtendedTimeoutCmd(PropertyCmd,
493
MillisecondsValueArgumentMixIn):
494
property = "ExtendedTimeout"
496
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
497
property = "Interval"
499
class SetApprovalDelayCmd(PropertyCmd,
500
MillisecondsValueArgumentMixIn):
501
property = "ApprovalDelay"
503
class SetApprovalDurationCmd(PropertyCmd,
504
MillisecondsValueArgumentMixIn):
505
property = "ApprovalDuration"
257
507
def has_actions(options):
258
508
return any((options.enable,
319
578
parser.add_argument("-s", "--secret",
320
579
type=argparse.FileType(mode="rb"),
321
580
help="Set password blob (file) for client")
322
parser.add_argument("-A", "--approve", action="store_true",
323
help="Approve any current client request")
324
parser.add_argument("-D", "--deny", action="store_true",
325
help="Deny any current client request")
581
approve_deny = parser.add_mutually_exclusive_group()
582
approve_deny.add_argument(
583
"-A", "--approve", action="store_true",
584
help="Approve any current client request")
585
approve_deny.add_argument("-D", "--deny", action="store_true",
586
help="Deny any current client request")
326
587
parser.add_argument("--check", action="store_true",
327
588
help="Run self-test")
328
589
parser.add_argument("client", nargs="*", help="Client name")
592
def commands_from_options(options):
596
if options.dump_json:
597
commands.append(DumpJSONCmd())
600
commands.append(EnableCmd())
603
commands.append(DisableCmd())
605
if options.bump_timeout:
606
commands.append(BumpTimeoutCmd())
608
if options.start_checker:
609
commands.append(StartCheckerCmd())
611
if options.stop_checker:
612
commands.append(StopCheckerCmd())
614
if options.is_enabled:
615
commands.append(IsEnabledCmd())
618
commands.append(RemoveCmd())
620
if options.checker is not None:
621
commands.append(SetCheckerCmd(options.checker))
623
if options.timeout is not None:
624
commands.append(SetTimeoutCmd(options.timeout))
626
if options.extended_timeout:
628
SetExtendedTimeoutCmd(options.extended_timeout))
630
if options.interval is not None:
631
commands.append(SetIntervalCmd(options.interval))
633
if options.approved_by_default is not None:
634
if options.approved_by_default:
635
commands.append(ApproveByDefaultCmd())
637
commands.append(DenyByDefaultCmd())
639
if options.approval_delay is not None:
640
commands.append(SetApprovalDelayCmd(options.approval_delay))
642
if options.approval_duration is not None:
644
SetApprovalDurationCmd(options.approval_duration))
646
if options.host is not None:
647
commands.append(SetHostCmd(options.host))
649
if options.secret is not None:
650
commands.append(SetSecretCmd(options.secret))
653
commands.append(ApproveCmd())
656
commands.append(DenyCmd())
658
# If no command option has been given, show table of clients,
659
# optionally verbosely
661
commands.append(PrintTableCmd(verbose=options.verbose))
667
parser = argparse.ArgumentParser()
669
add_command_line_options(parser)
329
671
options = parser.parse_args()
331
673
if has_actions(options) and not (options.client or options.all):
332
674
parser.error("Options require clients names or --all.")
333
675
if options.verbose and has_actions(options):
334
parser.error("--verbose can only be used alone or with"
676
parser.error("--verbose can only be used alone.")
677
if options.dump_json and (options.verbose
678
or has_actions(options)):
679
parser.error("--dump-json can only be used alone.")
336
680
if options.all and not has_actions(options):
337
681
parser.error("--all requires an action.")
340
fail_count, test_count = doctest.testmod()
341
sys.exit(os.EX_OK if fail_count == 0 else 1)
682
if options.is_enabled and len(options.client) > 1:
683
parser.error("--is-enabled requires exactly one client")
685
clientnames = options.client
344
688
bus = dbus.SystemBus()
345
689
mandos_dbus_objc = bus.get_object(busname, server_path)
346
690
except dbus.exceptions.DBusException:
347
print("Could not connect to Mandos server",
691
log.critical("Could not connect to Mandos server")
351
694
mandos_serv = dbus.Interface(mandos_dbus_objc,
352
dbus_interface = server_interface)
354
#block stderr since dbus library prints to stderr
355
null = os.open(os.path.devnull, os.O_RDWR)
356
stderrcopy = os.dup(sys.stderr.fileno())
357
os.dup2(null, sys.stderr.fileno())
695
dbus_interface=server_interface)
696
mandos_serv_object_manager = dbus.Interface(
697
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
699
# Filter out log message from dbus module
700
dbus_logger = logging.getLogger("dbus.proxies")
701
class NullFilter(logging.Filter):
702
def filter(self, record):
704
dbus_filter = NullFilter()
361
mandos_clients = mandos_serv.GetAllClientsWithProperties()
364
os.dup2(stderrcopy, sys.stderr.fileno())
366
except dbus.exceptions.DBusException:
367
print("Access denied: Accessing mandos server through dbus.",
706
dbus_logger.addFilter(dbus_filter)
707
mandos_clients = {path: ifs_and_props[client_interface]
708
for path, ifs_and_props in
709
mandos_serv_object_manager
710
.GetManagedObjects().items()
711
if client_interface in ifs_and_props}
712
except dbus.exceptions.DBusException as e:
713
log.critical("Failed to access Mandos server through D-Bus:"
717
# restore dbus logger
718
dbus_logger.removeFilter(dbus_filter)
371
720
# Compile dict of (clients: properties) to process
374
if options.all or not options.client:
375
clients = dict((bus.get_object(busname, path), properties)
376
for path, properties in
377
mandos_clients.items())
724
clients = {bus.get_object(busname, path): properties
725
for path, properties in mandos_clients.items()}
379
for name in options.client:
380
for path, client in mandos_clients.iteritems():
727
for name in clientnames:
728
for path, client in mandos_clients.items():
381
729
if client["Name"] == name:
382
730
client_objc = bus.get_object(busname, path)
383
731
clients[client_objc] = client
386
print("Client not found on server: {0!r}"
387
.format(name), file=sys.stderr)
734
log.critical("Client not found on server: %r", name)
390
if not has_actions(options) and clients:
392
keywords = ("Name", "Enabled", "Timeout",
393
"LastCheckedOK", "Created", "Interval",
394
"Host", "Fingerprint", "CheckerRunning",
395
"LastEnabled", "ApprovalPending",
397
"LastApprovalRequest", "ApprovalDelay",
398
"ApprovalDuration", "Checker",
401
keywords = defaultkeywords
403
print_clients(clients.values(), keywords)
405
# Process each client in the list by all selected options
406
for client in clients:
407
def set_client_prop(prop, value):
408
"""Set a Client D-Bus property"""
409
client.Set(client_interface, prop, value,
410
dbus_interface=dbus.PROPERTIES_IFACE)
411
def set_client_prop_ms(prop, value):
412
"""Set a Client D-Bus property, converted
413
from a string to milliseconds."""
414
set_client_prop(prop,
415
timedelta_to_milliseconds
416
(string_to_delta(value)))
418
mandos_serv.RemoveClient(client.__dbus_object_path__)
420
set_client_prop("Enabled", dbus.Boolean(True))
422
set_client_prop("Enabled", dbus.Boolean(False))
423
if options.bump_timeout:
424
set_client_prop("LastCheckedOK", "")
425
if options.start_checker:
426
set_client_prop("CheckerRunning", dbus.Boolean(True))
427
if options.stop_checker:
428
set_client_prop("CheckerRunning", dbus.Boolean(False))
429
if options.is_enabled:
430
sys.exit(0 if client.Get(client_interface,
433
dbus.PROPERTIES_IFACE)
435
if options.checker is not None:
436
set_client_prop("Checker", options.checker)
437
if options.host is not None:
438
set_client_prop("Host", options.host)
439
if options.interval is not None:
440
set_client_prop_ms("Interval", options.interval)
441
if options.approval_delay is not None:
442
set_client_prop_ms("ApprovalDelay",
443
options.approval_delay)
444
if options.approval_duration is not None:
445
set_client_prop_ms("ApprovalDuration",
446
options.approval_duration)
447
if options.timeout is not None:
448
set_client_prop_ms("Timeout", options.timeout)
449
if options.extended_timeout is not None:
450
set_client_prop_ms("ExtendedTimeout",
451
options.extended_timeout)
452
if options.secret is not None:
453
set_client_prop("Secret",
454
dbus.ByteArray(options.secret.read()))
455
if options.approved_by_default is not None:
456
set_client_prop("ApprovedByDefault",
458
.approved_by_default))
460
client.Approve(dbus.Boolean(True),
461
dbus_interface=client_interface)
463
client.Approve(dbus.Boolean(False),
464
dbus_interface=client_interface)
737
# Run all commands on clients
738
commands = commands_from_options(options)
739
for command in commands:
740
command.run(mandos_serv, clients)
743
class Test_milliseconds_to_string(unittest.TestCase):
745
self.assertEqual(milliseconds_to_string(93785000),
747
def test_no_days(self):
748
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
749
def test_all_zero(self):
750
self.assertEqual(milliseconds_to_string(0), "00:00:00")
751
def test_no_fractional_seconds(self):
752
self.assertEqual(milliseconds_to_string(400), "00:00:00")
753
self.assertEqual(milliseconds_to_string(900), "00:00:00")
754
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
756
class Test_string_to_delta(unittest.TestCase):
757
def test_handles_basic_rfc3339(self):
758
self.assertEqual(string_to_delta("PT0S"),
759
datetime.timedelta())
760
self.assertEqual(string_to_delta("P0D"),
761
datetime.timedelta())
762
self.assertEqual(string_to_delta("PT1S"),
763
datetime.timedelta(0, 1))
764
self.assertEqual(string_to_delta("PT2H"),
765
datetime.timedelta(0, 7200))
766
def test_falls_back_to_pre_1_6_1_with_warning(self):
767
# assertLogs only exists in Python 3.4
768
if hasattr(self, "assertLogs"):
769
with self.assertLogs(log, logging.WARNING):
770
value = string_to_delta("2h")
772
class WarningFilter(logging.Filter):
773
"""Don't show, but record the presence of, warnings"""
774
def filter(self, record):
775
is_warning = record.levelno >= logging.WARNING
776
self.found = is_warning or getattr(self, "found",
778
return not is_warning
779
warning_filter = WarningFilter()
780
log.addFilter(warning_filter)
782
value = string_to_delta("2h")
784
log.removeFilter(warning_filter)
785
self.assertTrue(getattr(warning_filter, "found", False))
786
self.assertEqual(value, datetime.timedelta(0, 7200))
789
class TestCmd(unittest.TestCase):
790
"""Abstract class for tests of command classes"""
793
class MockClient(object):
794
def __init__(self, name, **attributes):
795
self.__dbus_object_path__ = "objpath_{}".format(name)
796
self.attributes = attributes
797
self.attributes["Name"] = name
799
def Set(self, interface, property, value, dbus_interface):
800
testcase.assertEqual(interface, client_interface)
801
testcase.assertEqual(dbus_interface,
802
dbus.PROPERTIES_IFACE)
803
self.attributes[property] = value
804
def Get(self, interface, property, dbus_interface):
805
testcase.assertEqual(interface, client_interface)
806
testcase.assertEqual(dbus_interface,
807
dbus.PROPERTIES_IFACE)
808
return self.attributes[property]
809
def Approve(self, approve, dbus_interface):
810
testcase.assertEqual(dbus_interface, client_interface)
811
self.calls.append(("Approve", (approve,
813
self.client = MockClient(
815
KeyID=("92ed150794387c03ce684574b1139a65"
816
"94a34f895daaaf09fd8ea90a27cddb12"),
818
Host="foo.example.org",
819
Enabled=dbus.Boolean(True),
821
LastCheckedOK="2019-02-03T00:00:00",
822
Created="2019-01-02T00:00:00",
824
Fingerprint=("778827225BA7DE539C5A"
825
"7CFA59CFF7CDBD9A5920"),
826
CheckerRunning=dbus.Boolean(False),
827
LastEnabled="2019-01-03T00:00:00",
828
ApprovalPending=dbus.Boolean(False),
829
ApprovedByDefault=dbus.Boolean(True),
830
LastApprovalRequest="",
832
ApprovalDuration=1000,
833
Checker="fping -q -- %(host)s",
834
ExtendedTimeout=900000,
835
Expires="2019-02-04T00:00:00",
837
self.other_client = MockClient(
839
KeyID=("0558568eedd67d622f5c83b35a115f79"
840
"6ab612cff5ad227247e46c2b020f441c"),
843
Enabled=dbus.Boolean(True),
845
LastCheckedOK="2019-02-04T00:00:00",
846
Created="2019-01-03T00:00:00",
848
Fingerprint=("3E393AEAEFB84C7E89E2"
849
"F547B3A107558FCA3A27"),
850
CheckerRunning=dbus.Boolean(True),
851
LastEnabled="2019-01-04T00:00:00",
852
ApprovalPending=dbus.Boolean(False),
853
ApprovedByDefault=dbus.Boolean(False),
854
LastApprovalRequest="2019-01-03T00:00:00",
856
ApprovalDuration=1000,
858
ExtendedTimeout=900000,
859
Expires="2019-02-05T00:00:00",
860
LastCheckerStatus=-2)
861
self.clients = collections.OrderedDict(
863
(self.client, self.client.attributes),
864
(self.other_client, self.other_client.attributes),
866
self.one_client = {self.client: self.client.attributes}
868
class TestPrintTableCmd(TestCmd):
869
def test_normal(self):
870
output = PrintTableCmd().output(self.clients)
871
expected_output = """
872
Name Enabled Timeout Last Successful Check
873
foo Yes 00:05:00 2019-02-03T00:00:00
874
barbar Yes 00:05:00 2019-02-04T00:00:00
876
self.assertEqual(output, expected_output)
877
def test_verbose(self):
878
output = PrintTableCmd(verbose=True).output(self.clients)
879
expected_output = """
880
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
881
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
882
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
884
self.assertEqual(output, expected_output)
885
def test_one_client(self):
886
output = PrintTableCmd().output(self.one_client)
887
expected_output = """
888
Name Enabled Timeout Last Successful Check
889
foo Yes 00:05:00 2019-02-03T00:00:00
891
self.assertEqual(output, expected_output)
893
class TestDumpJSONCmd(TestCmd):
895
self.expected_json = {
898
"KeyID": ("92ed150794387c03ce684574b1139a65"
899
"94a34f895daaaf09fd8ea90a27cddb12"),
900
"Host": "foo.example.org",
903
"LastCheckedOK": "2019-02-03T00:00:00",
904
"Created": "2019-01-02T00:00:00",
906
"Fingerprint": ("778827225BA7DE539C5A"
907
"7CFA59CFF7CDBD9A5920"),
908
"CheckerRunning": False,
909
"LastEnabled": "2019-01-03T00:00:00",
910
"ApprovalPending": False,
911
"ApprovedByDefault": True,
912
"LastApprovalRequest": "",
914
"ApprovalDuration": 1000,
915
"Checker": "fping -q -- %(host)s",
916
"ExtendedTimeout": 900000,
917
"Expires": "2019-02-04T00:00:00",
918
"LastCheckerStatus": 0,
922
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
923
"6ab612cff5ad227247e46c2b020f441c"),
927
"LastCheckedOK": "2019-02-04T00:00:00",
928
"Created": "2019-01-03T00:00:00",
930
"Fingerprint": ("3E393AEAEFB84C7E89E2"
931
"F547B3A107558FCA3A27"),
932
"CheckerRunning": True,
933
"LastEnabled": "2019-01-04T00:00:00",
934
"ApprovalPending": False,
935
"ApprovedByDefault": False,
936
"LastApprovalRequest": "2019-01-03T00:00:00",
937
"ApprovalDelay": 30000,
938
"ApprovalDuration": 1000,
940
"ExtendedTimeout": 900000,
941
"Expires": "2019-02-05T00:00:00",
942
"LastCheckerStatus": -2,
945
return super(TestDumpJSONCmd, self).setUp()
946
def test_normal(self):
947
json_data = json.loads(DumpJSONCmd().output(self.clients))
948
self.assertDictEqual(json_data, self.expected_json)
949
def test_one_client(self):
950
clients = self.one_client
951
json_data = json.loads(DumpJSONCmd().output(clients))
952
expected_json = {"foo": self.expected_json["foo"]}
953
self.assertDictEqual(json_data, expected_json)
955
class TestIsEnabledCmd(TestCmd):
956
def test_is_enabled(self):
957
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
958
for client, properties in self.clients.items()))
959
def test_is_enabled_run_exits_successfully(self):
960
with self.assertRaises(SystemExit) as e:
961
IsEnabledCmd().run(None, self.one_client)
962
if e.exception.code is not None:
963
self.assertEqual(e.exception.code, 0)
965
self.assertIsNone(e.exception.code)
966
def test_is_enabled_run_exits_with_failure(self):
967
self.client.attributes["Enabled"] = dbus.Boolean(False)
968
with self.assertRaises(SystemExit) as e:
969
IsEnabledCmd().run(None, self.one_client)
970
if isinstance(e.exception.code, int):
971
self.assertNotEqual(e.exception.code, 0)
973
self.assertIsNotNone(e.exception.code)
975
class TestRemoveCmd(TestCmd):
976
def test_remove(self):
977
class MockMandos(object):
980
def RemoveClient(self, dbus_path):
981
self.calls.append(("RemoveClient", (dbus_path,)))
982
mandos = MockMandos()
983
super(TestRemoveCmd, self).setUp()
984
RemoveCmd().run(mandos, self.clients)
985
self.assertEqual(len(mandos.calls), 2)
986
for client in self.clients:
987
self.assertIn(("RemoveClient",
988
(client.__dbus_object_path__,)),
991
class TestApproveCmd(TestCmd):
992
def test_approve(self):
993
ApproveCmd().run(None, self.clients)
994
for client in self.clients:
995
self.assertIn(("Approve", (True, client_interface)),
998
class TestDenyCmd(TestCmd):
1000
DenyCmd().run(None, self.clients)
1001
for client in self.clients:
1002
self.assertIn(("Approve", (False, client_interface)),
1005
class TestEnableCmd(TestCmd):
1006
def test_enable(self):
1007
for client in self.clients:
1008
client.attributes["Enabled"] = False
1010
EnableCmd().run(None, self.clients)
1012
for client in self.clients:
1013
self.assertTrue(client.attributes["Enabled"])
1015
class TestDisableCmd(TestCmd):
1016
def test_disable(self):
1017
DisableCmd().run(None, self.clients)
1019
for client in self.clients:
1020
self.assertFalse(client.attributes["Enabled"])
1022
class Unique(object):
1023
"""Class for objects which exist only to be unique objects, since
1024
unittest.mock.sentinel only exists in Python 3.3"""
1026
class TestPropertyCmd(TestCmd):
1027
"""Abstract class for tests of PropertyCmd classes"""
1029
if not hasattr(self, "command"):
1031
values_to_get = getattr(self, "values_to_get",
1033
for value_to_set, value_to_get in zip(self.values_to_set,
1035
for client in self.clients:
1036
old_value = client.attributes[self.property]
1037
self.assertNotIsInstance(old_value, Unique)
1038
client.attributes[self.property] = Unique()
1039
self.run_command(value_to_set, self.clients)
1040
for client in self.clients:
1041
value = client.attributes[self.property]
1042
self.assertNotIsInstance(value, Unique)
1043
self.assertEqual(value, value_to_get)
1044
def run_command(self, value, clients):
1045
self.command().run(None, clients)
1047
class TestBumpTimeoutCmd(TestPropertyCmd):
1048
command = BumpTimeoutCmd
1049
property = "LastCheckedOK"
1050
values_to_set = [""]
1052
class TestStartCheckerCmd(TestPropertyCmd):
1053
command = StartCheckerCmd
1054
property = "CheckerRunning"
1055
values_to_set = [dbus.Boolean(True)]
1057
class TestStopCheckerCmd(TestPropertyCmd):
1058
command = StopCheckerCmd
1059
property = "CheckerRunning"
1060
values_to_set = [dbus.Boolean(False)]
1062
class TestApproveByDefaultCmd(TestPropertyCmd):
1063
command = ApproveByDefaultCmd
1064
property = "ApprovedByDefault"
1065
values_to_set = [dbus.Boolean(True)]
1067
class TestDenyByDefaultCmd(TestPropertyCmd):
1068
command = DenyByDefaultCmd
1069
property = "ApprovedByDefault"
1070
values_to_set = [dbus.Boolean(False)]
1072
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1073
"""Abstract class for tests of PropertyCmd classes using the
1074
ValueArgumentMixIn"""
1076
if type(self) is TestValueArgumentPropertyCmd:
1078
return super(TestValueArgumentPropertyCmd, self).runTest()
1079
def run_command(self, value, clients):
1080
self.command(value).run(None, clients)
1082
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1083
command = SetCheckerCmd
1084
property = "Checker"
1085
values_to_set = ["", ":", "fping -q -- %s"]
1087
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1088
command = SetHostCmd
1090
values_to_set = ["192.0.2.3", "foo.example.org"]
1092
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1093
command = SetSecretCmd
1095
values_to_set = [open("/dev/null", "rb"),
1096
io.BytesIO(b"secret\0xyzzy\nbar")]
1097
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1099
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1100
command = SetTimeoutCmd
1101
property = "Timeout"
1102
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1103
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1105
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1106
command = SetExtendedTimeoutCmd
1107
property = "ExtendedTimeout"
1108
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1109
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1111
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1112
command = SetIntervalCmd
1113
property = "Interval"
1114
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1115
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1117
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1118
command = SetApprovalDelayCmd
1119
property = "ApprovalDelay"
1120
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1121
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1123
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1124
command = SetApprovalDurationCmd
1125
property = "ApprovalDuration"
1126
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1127
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1129
class TestOptions(unittest.TestCase):
1131
self.parser = argparse.ArgumentParser()
1132
add_command_line_options(self.parser)
1133
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1134
"""Assert that parsing ARGS should result in an instance of
1135
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1136
options = self.parser.parse_args(args)
1137
commands = commands_from_options(options)
1138
self.assertEqual(len(commands), 1)
1139
command = commands[0]
1140
self.assertIsInstance(command, command_cls)
1141
for key, value in cmd_attrs.items():
1142
self.assertEqual(getattr(command, key), value)
1143
def test_default_is_show_table(self):
1144
self.assert_command_from_args([], PrintTableCmd,
1146
def test_show_table_verbose(self):
1147
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1149
def test_enable(self):
1150
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1151
def test_disable(self):
1152
self.assert_command_from_args(["--disable", "foo"],
1157
def should_only_run_tests():
1158
parser = argparse.ArgumentParser(add_help=False)
1159
parser.add_argument("--check", action='store_true')
1160
args, unknown_args = parser.parse_known_args()
1161
run_tests = args.check
1163
# Remove --check argument from sys.argv
1164
sys.argv[1:] = unknown_args
1167
# Add all tests from doctest strings
1168
def load_tests(loader, tests, none):
1170
tests.addTests(doctest.DocTestSuite())
466
1173
if __name__ == "__main__":
1174
if should_only_run_tests():
1175
# Call using ./tdd-python-script --check [--verbose]