77
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
80
def milliseconds_to_string(ms):
81
td = datetime.timedelta(0, 0, 0, ms)
82
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
83
.format(days="{}T".format(td.days) if td.days else "",
84
hours=td.seconds // 3600,
85
minutes=(td.seconds % 3600) // 60,
86
seconds=td.seconds % 60))
83
parser = argparse.ArgumentParser()
85
add_command_line_options(parser)
87
options = parser.parse_args()
89
check_option_syntax(parser, options)
91
clientnames = options.client
94
log.setLevel(logging.DEBUG)
96
bus = dbus.SystemBus()
98
mandos_dbus_object = get_mandos_dbus_object(bus)
100
mandos_serv = dbus.Interface(
101
mandos_dbus_object, dbus_interface=server_dbus_interface)
102
mandos_serv_object_manager = dbus.Interface(
103
mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
105
managed_objects = get_managed_objects(mandos_serv_object_manager)
108
for path, ifs_and_props in managed_objects.items():
110
all_clients[path] = ifs_and_props[client_dbus_interface]
114
# Compile dict of (clientpath: properties) to process
116
clients = all_clients
119
for name in clientnames:
120
for objpath, properties in all_clients.items():
121
if properties["Name"] == name:
122
clients[objpath] = properties
125
log.critical("Client not found on server: %r", name)
128
# Run all commands on clients
129
commands = commands_from_options(options)
130
for command in commands:
131
command.run(clients, bus, mandos_serv)
134
def add_command_line_options(parser):
135
parser.add_argument("--version", action="version",
136
version="%(prog)s {}".format(version),
137
help="show version number and exit")
138
parser.add_argument("-a", "--all", action="store_true",
139
help="Select all clients")
140
parser.add_argument("-v", "--verbose", action="store_true",
141
help="Print all fields")
142
parser.add_argument("-j", "--dump-json", action="store_true",
143
help="Dump client data in JSON format")
144
enable_disable = parser.add_mutually_exclusive_group()
145
enable_disable.add_argument("-e", "--enable", action="store_true",
146
help="Enable client")
147
enable_disable.add_argument("-d", "--disable",
149
help="disable client")
150
parser.add_argument("-b", "--bump-timeout", action="store_true",
151
help="Bump timeout for client")
152
start_stop_checker = parser.add_mutually_exclusive_group()
153
start_stop_checker.add_argument("--start-checker",
155
help="Start checker for client")
156
start_stop_checker.add_argument("--stop-checker",
158
help="Stop checker for client")
159
parser.add_argument("-V", "--is-enabled", action="store_true",
160
help="Check if client is enabled")
161
parser.add_argument("-r", "--remove", action="store_true",
162
help="Remove client")
163
parser.add_argument("-c", "--checker",
164
help="Set checker command for client")
165
parser.add_argument("-t", "--timeout", type=string_to_delta,
166
help="Set timeout for client")
167
parser.add_argument("--extended-timeout", type=string_to_delta,
168
help="Set extended timeout for client")
169
parser.add_argument("-i", "--interval", type=string_to_delta,
170
help="Set checker interval for client")
171
approve_deny_default = parser.add_mutually_exclusive_group()
172
approve_deny_default.add_argument(
173
"--approve-by-default", action="store_true",
174
default=None, dest="approved_by_default",
175
help="Set client to be approved by default")
176
approve_deny_default.add_argument(
177
"--deny-by-default", action="store_false",
178
dest="approved_by_default",
179
help="Set client to be denied by default")
180
parser.add_argument("--approval-delay", type=string_to_delta,
181
help="Set delay before client approve/deny")
182
parser.add_argument("--approval-duration", type=string_to_delta,
183
help="Set duration of one client approval")
184
parser.add_argument("-H", "--host", help="Set host for client")
185
parser.add_argument("-s", "--secret",
186
type=argparse.FileType(mode="rb"),
187
help="Set password blob (file) for client")
188
approve_deny = parser.add_mutually_exclusive_group()
189
approve_deny.add_argument(
190
"-A", "--approve", action="store_true",
191
help="Approve any current client request")
192
approve_deny.add_argument("-D", "--deny", action="store_true",
193
help="Deny any current client request")
194
parser.add_argument("--debug", action="store_true",
195
help="Debug mode (show D-Bus commands)")
196
parser.add_argument("--check", action="store_true",
197
help="Run self-test")
198
parser.add_argument("client", nargs="*", help="Client name")
201
def string_to_delta(interval):
202
"""Parse a string and return a datetime.timedelta"""
205
return rfc3339_duration_to_delta(interval)
206
except ValueError as e:
207
log.warning("%s - Parsing as pre-1.6.1 interval instead",
209
return parse_pre_1_6_1_interval(interval)
89
212
def rfc3339_duration_to_delta(duration):
273
## Classes for commands.
275
# Abstract classes first
385
def check_option_syntax(parser, options):
386
"""Apply additional restrictions on options, not expressible in
389
def has_actions(options):
390
return any((options.enable,
392
options.bump_timeout,
393
options.start_checker,
394
options.stop_checker,
397
options.checker is not None,
398
options.timeout is not None,
399
options.extended_timeout is not None,
400
options.interval is not None,
401
options.approved_by_default is not None,
402
options.approval_delay is not None,
403
options.approval_duration is not None,
404
options.host is not None,
405
options.secret is not None,
409
if has_actions(options) and not (options.client or options.all):
410
parser.error("Options require clients names or --all.")
411
if options.verbose and has_actions(options):
412
parser.error("--verbose can only be used alone.")
413
if options.dump_json and (options.verbose
414
or has_actions(options)):
415
parser.error("--dump-json can only be used alone.")
416
if options.all and not has_actions(options):
417
parser.error("--all requires an action.")
418
if options.is_enabled and len(options.client) > 1:
419
parser.error("--is-enabled requires exactly one client")
421
options.remove = False
422
if has_actions(options) and not options.deny:
423
parser.error("--remove can only be combined with --deny")
424
options.remove = True
427
def get_mandos_dbus_object(bus):
428
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
429
dbus_busname, server_dbus_path)
430
with if_dbus_exception_log_with_exception_and_exit(
431
"Could not connect to Mandos server: %s"):
432
mandos_dbus_object = bus.get_object(dbus_busname,
434
return mandos_dbus_object
437
@contextlib.contextmanager
438
def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):
441
except dbus.exceptions.DBusException as e:
442
log.critical(*(args + (e,)), **kwargs)
446
def get_managed_objects(object_manager):
447
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
448
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
449
with if_dbus_exception_log_with_exception_and_exit(
450
"Failed to access Mandos server through D-Bus:\n%s"):
451
with SilenceLogger("dbus.proxies"):
452
managed_objects = object_manager.GetManagedObjects()
453
return managed_objects
456
class SilenceLogger(object):
457
"Simple context manager to silence a particular logger"
458
def __init__(self, loggername):
459
self.logger = logging.getLogger(loggername)
462
self.logger.addFilter(self.nullfilter)
465
class NullFilter(logging.Filter):
466
def filter(self, record):
469
nullfilter = NullFilter()
471
def __exit__(self, exc_type, exc_val, exc_tb):
472
self.logger.removeFilter(self.nullfilter)
475
def commands_from_options(options):
479
if options.is_enabled:
480
commands.append(IsEnabledCmd())
483
commands.append(ApproveCmd())
486
commands.append(DenyCmd())
489
commands.append(RemoveCmd())
491
if options.dump_json:
492
commands.append(DumpJSONCmd())
495
commands.append(EnableCmd())
498
commands.append(DisableCmd())
500
if options.bump_timeout:
501
commands.append(BumpTimeoutCmd())
503
if options.start_checker:
504
commands.append(StartCheckerCmd())
506
if options.stop_checker:
507
commands.append(StopCheckerCmd())
509
if options.approved_by_default is not None:
510
if options.approved_by_default:
511
commands.append(ApproveByDefaultCmd())
513
commands.append(DenyByDefaultCmd())
515
if options.checker is not None:
516
commands.append(SetCheckerCmd(options.checker))
518
if options.host is not None:
519
commands.append(SetHostCmd(options.host))
521
if options.secret is not None:
522
commands.append(SetSecretCmd(options.secret))
524
if options.timeout is not None:
525
commands.append(SetTimeoutCmd(options.timeout))
527
if options.extended_timeout:
529
SetExtendedTimeoutCmd(options.extended_timeout))
531
if options.interval is not None:
532
commands.append(SetIntervalCmd(options.interval))
534
if options.approval_delay is not None:
535
commands.append(SetApprovalDelayCmd(options.approval_delay))
537
if options.approval_duration is not None:
539
SetApprovalDurationCmd(options.approval_duration))
541
# If no command option has been given, show table of clients,
542
# optionally verbosely
544
commands.append(PrintTableCmd(verbose=options.verbose))
276
549
class Command(object):
277
550
"""Abstract class for commands"""
278
def run(self, mandos, clients):
551
def run(self, clients, bus=None, mandos=None):
279
552
"""Normal commands should implement run_on_one_client(), but
280
553
commands which want to operate on all clients at the same time
281
554
can override this run() method instead."""
282
555
self.mandos = mandos
283
for client, properties in clients.items():
556
for clientpath, properties in clients.items():
557
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
558
dbus_busname, str(clientpath))
559
client = bus.get_object(dbus_busname, clientpath)
284
560
self.run_on_one_client(client, properties)
286
class PrintCmd(Command):
287
"""Abstract class for commands printing client details"""
563
class IsEnabledCmd(Command):
564
def run(self, clients, bus=None, mandos=None):
565
client, properties = next(iter(clients.items()))
566
if self.is_enabled(client, properties):
569
def is_enabled(self, client, properties):
570
return properties["Enabled"]
573
class ApproveCmd(Command):
574
def run_on_one_client(self, client, properties):
575
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
576
client.__dbus_object_path__, client_dbus_interface)
577
client.Approve(dbus.Boolean(True),
578
dbus_interface=client_dbus_interface)
581
class DenyCmd(Command):
582
def run_on_one_client(self, client, properties):
583
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
584
client.__dbus_object_path__, client_dbus_interface)
585
client.Approve(dbus.Boolean(False),
586
dbus_interface=client_dbus_interface)
589
class RemoveCmd(Command):
590
def run_on_one_client(self, client, properties):
591
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
592
server_dbus_path, server_dbus_interface,
593
str(client.__dbus_object_path__))
594
self.mandos.RemoveClient(client.__dbus_object_path__)
597
class OutputCmd(Command):
598
"""Abstract class for commands outputting client details"""
288
599
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
289
600
"Created", "Interval", "Host", "KeyID",
290
601
"Fingerprint", "CheckerRunning", "LastEnabled",
405
710
**{key: self.string_from_client(client, key)
406
711
for key in self.keywords})
410
class DumpJSONCmd(PrintCmd):
411
def output(self, clients):
412
data = {client["Name"]:
413
{key: self.dbus_boolean_to_bool(client[key])
414
for key in self.all_keywords}
415
for client in clients.values()}
416
return json.dumps(data, indent=4, separators=(',', ': '))
418
def dbus_boolean_to_bool(value):
419
if isinstance(value, dbus.Boolean):
423
class IsEnabledCmd(Command):
424
def run_on_one_client(self, client, properties):
425
if self.is_enabled(client, properties):
428
def is_enabled(self, client, properties):
429
return bool(properties["Enabled"])
431
class RemoveCmd(Command):
432
def run_on_one_client(self, client, properties):
433
self.mandos.RemoveClient(client.__dbus_object_path__)
435
class ApproveCmd(Command):
436
def run_on_one_client(self, client, properties):
437
client.Approve(dbus.Boolean(True),
438
dbus_interface=client_interface)
440
class DenyCmd(Command):
441
def run_on_one_client(self, client, properties):
442
client.Approve(dbus.Boolean(False),
443
dbus_interface=client_interface)
714
def milliseconds_to_string(ms):
715
td = datetime.timedelta(0, 0, 0, ms)
716
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
717
.format(days="{}T".format(td.days)
719
hours=td.seconds // 3600,
720
minutes=(td.seconds % 3600) // 60,
721
seconds=td.seconds % 60))
724
class PropertyCmd(Command):
725
"""Abstract class for Actions for setting one client property"""
727
def run_on_one_client(self, client, properties):
728
"""Set the Client's D-Bus property"""
729
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
730
client.__dbus_object_path__,
731
dbus.PROPERTIES_IFACE, client_dbus_interface,
732
self.propname, self.value_to_set
733
if not isinstance(self.value_to_set, dbus.Boolean)
734
else bool(self.value_to_set))
735
client.Set(client_dbus_interface, self.propname,
737
dbus_interface=dbus.PROPERTIES_IFACE)
741
raise NotImplementedError()
445
744
class EnableCmd(PropertyCmd):
447
746
value_to_set = dbus.Boolean(True)
449
749
class DisableCmd(PropertyCmd):
451
751
value_to_set = dbus.Boolean(False)
453
754
class BumpTimeoutCmd(PropertyCmd):
454
property = "LastCheckedOK"
755
propname = "LastCheckedOK"
455
756
value_to_set = ""
457
759
class StartCheckerCmd(PropertyCmd):
458
property = "CheckerRunning"
760
propname = "CheckerRunning"
459
761
value_to_set = dbus.Boolean(True)
461
764
class StopCheckerCmd(PropertyCmd):
462
property = "CheckerRunning"
765
propname = "CheckerRunning"
463
766
value_to_set = dbus.Boolean(False)
465
769
class ApproveByDefaultCmd(PropertyCmd):
466
property = "ApprovedByDefault"
770
propname = "ApprovedByDefault"
467
771
value_to_set = dbus.Boolean(True)
469
774
class DenyByDefaultCmd(PropertyCmd):
470
property = "ApprovedByDefault"
775
propname = "ApprovedByDefault"
471
776
value_to_set = dbus.Boolean(False)
473
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
476
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
479
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
779
class PropertyValueCmd(PropertyCmd):
780
"""Abstract class for PropertyCmd recieving a value as argument"""
781
def __init__(self, value):
782
self.value_to_set = value
785
class SetCheckerCmd(PropertyValueCmd):
789
class SetHostCmd(PropertyValueCmd):
793
class SetSecretCmd(PropertyValueCmd):
481
797
def value_to_set(self):
483
800
@value_to_set.setter
484
801
def value_to_set(self, value):
485
802
"""When setting, read data from supplied file object"""
486
803
self._vts = value.read()
490
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
493
class SetExtendedTimeoutCmd(PropertyCmd,
494
MillisecondsValueArgumentMixIn):
495
property = "ExtendedTimeout"
497
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
498
property = "Interval"
500
class SetApprovalDelayCmd(PropertyCmd,
501
MillisecondsValueArgumentMixIn):
502
property = "ApprovalDelay"
504
class SetApprovalDurationCmd(PropertyCmd,
505
MillisecondsValueArgumentMixIn):
506
property = "ApprovalDuration"
508
def has_actions(options):
509
return any((options.enable,
511
options.bump_timeout,
512
options.start_checker,
513
options.stop_checker,
516
options.checker is not None,
517
options.timeout is not None,
518
options.extended_timeout is not None,
519
options.interval is not None,
520
options.approved_by_default is not None,
521
options.approval_delay is not None,
522
options.approval_duration is not None,
523
options.host is not None,
524
options.secret is not None,
528
def add_command_line_options(parser):
529
parser.add_argument("--version", action="version",
530
version="%(prog)s {}".format(version),
531
help="show version number and exit")
532
parser.add_argument("-a", "--all", action="store_true",
533
help="Select all clients")
534
parser.add_argument("-v", "--verbose", action="store_true",
535
help="Print all fields")
536
parser.add_argument("-j", "--dump-json", action="store_true",
537
help="Dump client data in JSON format")
538
enable_disable = parser.add_mutually_exclusive_group()
539
enable_disable.add_argument("-e", "--enable", action="store_true",
540
help="Enable client")
541
enable_disable.add_argument("-d", "--disable",
543
help="disable client")
544
parser.add_argument("-b", "--bump-timeout", action="store_true",
545
help="Bump timeout for client")
546
start_stop_checker = parser.add_mutually_exclusive_group()
547
start_stop_checker.add_argument("--start-checker",
549
help="Start checker for client")
550
start_stop_checker.add_argument("--stop-checker",
552
help="Stop checker for client")
553
parser.add_argument("-V", "--is-enabled", action="store_true",
554
help="Check if client is enabled")
555
parser.add_argument("-r", "--remove", action="store_true",
556
help="Remove client")
557
parser.add_argument("-c", "--checker",
558
help="Set checker command for client")
559
parser.add_argument("-t", "--timeout",
560
help="Set timeout for client")
561
parser.add_argument("--extended-timeout",
562
help="Set extended timeout for client")
563
parser.add_argument("-i", "--interval",
564
help="Set checker interval for client")
565
approve_deny_default = parser.add_mutually_exclusive_group()
566
approve_deny_default.add_argument(
567
"--approve-by-default", action="store_true",
568
default=None, dest="approved_by_default",
569
help="Set client to be approved by default")
570
approve_deny_default.add_argument(
571
"--deny-by-default", action="store_false",
572
dest="approved_by_default",
573
help="Set client to be denied by default")
574
parser.add_argument("--approval-delay",
575
help="Set delay before client approve/deny")
576
parser.add_argument("--approval-duration",
577
help="Set duration of one client approval")
578
parser.add_argument("-H", "--host", help="Set host for client")
579
parser.add_argument("-s", "--secret",
580
type=argparse.FileType(mode="rb"),
581
help="Set password blob (file) for client")
582
approve_deny = parser.add_mutually_exclusive_group()
583
approve_deny.add_argument(
584
"-A", "--approve", action="store_true",
585
help="Approve any current client request")
586
approve_deny.add_argument("-D", "--deny", action="store_true",
587
help="Deny any current client request")
588
parser.add_argument("--check", action="store_true",
589
help="Run self-test")
590
parser.add_argument("client", nargs="*", help="Client name")
593
def commands_from_options(options):
597
if options.dump_json:
598
commands.append(DumpJSONCmd())
601
commands.append(EnableCmd())
604
commands.append(DisableCmd())
606
if options.bump_timeout:
607
commands.append(BumpTimeoutCmd())
609
if options.start_checker:
610
commands.append(StartCheckerCmd())
612
if options.stop_checker:
613
commands.append(StopCheckerCmd())
615
if options.is_enabled:
616
commands.append(IsEnabledCmd())
619
commands.append(RemoveCmd())
621
if options.checker is not None:
622
commands.append(SetCheckerCmd(options.checker))
624
if options.timeout is not None:
625
commands.append(SetTimeoutCmd(options.timeout))
627
if options.extended_timeout:
629
SetExtendedTimeoutCmd(options.extended_timeout))
631
if options.interval is not None:
632
commands.append(SetIntervalCmd(options.interval))
634
if options.approved_by_default is not None:
635
if options.approved_by_default:
636
commands.append(ApproveByDefaultCmd())
638
commands.append(DenyByDefaultCmd())
640
if options.approval_delay is not None:
641
commands.append(SetApprovalDelayCmd(options.approval_delay))
643
if options.approval_duration is not None:
645
SetApprovalDurationCmd(options.approval_duration))
647
if options.host is not None:
648
commands.append(SetHostCmd(options.host))
650
if options.secret is not None:
651
commands.append(SetSecretCmd(options.secret))
654
commands.append(ApproveCmd())
657
commands.append(DenyCmd())
659
# If no command option has been given, show table of clients,
660
# optionally verbosely
662
commands.append(PrintTableCmd(verbose=options.verbose))
668
parser = argparse.ArgumentParser()
670
add_command_line_options(parser)
672
options = parser.parse_args()
674
if has_actions(options) and not (options.client or options.all):
675
parser.error("Options require clients names or --all.")
676
if options.verbose and has_actions(options):
677
parser.error("--verbose can only be used alone.")
678
if options.dump_json and (options.verbose
679
or has_actions(options)):
680
parser.error("--dump-json can only be used alone.")
681
if options.all and not has_actions(options):
682
parser.error("--all requires an action.")
683
if options.is_enabled and len(options.client) > 1:
684
parser.error("--is-enabled requires exactly one client")
686
clientnames = options.client
689
bus = dbus.SystemBus()
690
mandos_dbus_objc = bus.get_object(busname, server_path)
691
except dbus.exceptions.DBusException:
692
log.critical("Could not connect to Mandos server")
695
mandos_serv = dbus.Interface(mandos_dbus_objc,
696
dbus_interface=server_interface)
697
mandos_serv_object_manager = dbus.Interface(
698
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
700
# Filter out log message from dbus module
701
dbus_logger = logging.getLogger("dbus.proxies")
702
class NullFilter(logging.Filter):
703
def filter(self, record):
705
dbus_filter = NullFilter()
707
dbus_logger.addFilter(dbus_filter)
708
mandos_clients = {path: ifs_and_props[client_interface]
709
for path, ifs_and_props in
710
mandos_serv_object_manager
711
.GetManagedObjects().items()
712
if client_interface in ifs_and_props}
713
except dbus.exceptions.DBusException as e:
714
log.critical("Failed to access Mandos server through D-Bus:"
718
# restore dbus logger
719
dbus_logger.removeFilter(dbus_filter)
721
# Compile dict of (clients: properties) to process
725
clients = {bus.get_object(busname, path): properties
726
for path, properties in mandos_clients.items()}
728
for name in clientnames:
729
for path, client in mandos_clients.items():
730
if client["Name"] == name:
731
client_objc = bus.get_object(busname, path)
732
clients[client_objc] = client
735
log.critical("Client not found on server: %r", name)
738
# Run all commands on clients
739
commands = commands_from_options(options)
740
for command in commands:
741
command.run(mandos_serv, clients)
807
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
808
"""Abstract class for PropertyValueCmd taking a value argument as
809
a datetime.timedelta() but should store it as milliseconds."""
812
def value_to_set(self):
816
def value_to_set(self, value):
817
"""When setting, convert value from a datetime.timedelta"""
818
self._vts = int(round(value.total_seconds() * 1000))
821
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
825
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
826
propname = "ExtendedTimeout"
829
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
830
propname = "Interval"
833
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
834
propname = "ApprovalDelay"
837
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
838
propname = "ApprovalDuration"
744
class Test_milliseconds_to_string(unittest.TestCase):
746
self.assertEqual(milliseconds_to_string(93785000),
748
def test_no_days(self):
749
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
750
def test_all_zero(self):
751
self.assertEqual(milliseconds_to_string(0), "00:00:00")
752
def test_no_fractional_seconds(self):
753
self.assertEqual(milliseconds_to_string(400), "00:00:00")
754
self.assertEqual(milliseconds_to_string(900), "00:00:00")
755
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
757
class Test_string_to_delta(unittest.TestCase):
842
class TestCaseWithAssertLogs(unittest.TestCase):
843
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
845
if not hasattr(unittest.TestCase, "assertLogs"):
846
@contextlib.contextmanager
847
def assertLogs(self, logger, level=logging.INFO):
848
capturing_handler = self.CapturingLevelHandler(level)
849
old_level = logger.level
850
old_propagate = logger.propagate
851
logger.addHandler(capturing_handler)
852
logger.setLevel(level)
853
logger.propagate = False
855
yield capturing_handler.watcher
857
logger.propagate = old_propagate
858
logger.removeHandler(capturing_handler)
859
logger.setLevel(old_level)
860
self.assertGreater(len(capturing_handler.watcher.records),
863
class CapturingLevelHandler(logging.Handler):
864
def __init__(self, level, *args, **kwargs):
865
logging.Handler.__init__(self, *args, **kwargs)
866
self.watcher = self.LoggingWatcher([], [])
867
def emit(self, record):
868
self.watcher.records.append(record)
869
self.watcher.output.append(self.format(record))
871
LoggingWatcher = collections.namedtuple("LoggingWatcher",
875
class Test_string_to_delta(TestCaseWithAssertLogs):
758
876
def test_handles_basic_rfc3339(self):
759
877
self.assertEqual(string_to_delta("PT0S"),
760
878
datetime.timedelta())
764
882
datetime.timedelta(0, 1))
765
883
self.assertEqual(string_to_delta("PT2H"),
766
884
datetime.timedelta(0, 7200))
767
886
def test_falls_back_to_pre_1_6_1_with_warning(self):
768
# assertLogs only exists in Python 3.4
769
if hasattr(self, "assertLogs"):
770
with self.assertLogs(log, logging.WARNING):
771
value = string_to_delta("2h")
773
class WarningFilter(logging.Filter):
774
"""Don't show, but record the presence of, warnings"""
775
def filter(self, record):
776
is_warning = record.levelno >= logging.WARNING
777
self.found = is_warning or getattr(self, "found",
779
return not is_warning
780
warning_filter = WarningFilter()
781
log.addFilter(warning_filter)
783
value = string_to_delta("2h")
785
log.removeFilter(warning_filter)
786
self.assertTrue(getattr(warning_filter, "found", False))
887
with self.assertLogs(log, logging.WARNING):
888
value = string_to_delta("2h")
787
889
self.assertEqual(value, datetime.timedelta(0, 7200))
892
class Test_check_option_syntax(unittest.TestCase):
894
self.parser = argparse.ArgumentParser()
895
add_command_line_options(self.parser)
897
def test_actions_requires_client_or_all(self):
898
for action, value in self.actions.items():
899
options = self.parser.parse_args()
900
setattr(options, action, value)
901
with self.assertParseError():
902
self.check_option_syntax(options)
904
# This mostly corresponds to the definition from has_actions() in
905
# check_option_syntax()
907
# The actual values set here are not that important, but we do
908
# at least stick to the correct types, even though they are
912
"bump_timeout": True,
913
"start_checker": True,
914
"stop_checker": True,
918
"timeout": datetime.timedelta(),
919
"extended_timeout": datetime.timedelta(),
920
"interval": datetime.timedelta(),
921
"approved_by_default": True,
922
"approval_delay": datetime.timedelta(),
923
"approval_duration": datetime.timedelta(),
925
"secret": io.BytesIO(b"x"),
930
@contextlib.contextmanager
931
def assertParseError(self):
932
with self.assertRaises(SystemExit) as e:
933
with self.temporarily_suppress_stderr():
935
# Exit code from argparse is guaranteed to be "2". Reference:
936
# https://docs.python.org/3/library
937
# /argparse.html#exiting-methods
938
self.assertEqual(e.exception.code, 2)
941
@contextlib.contextmanager
942
def temporarily_suppress_stderr():
943
null = os.open(os.path.devnull, os.O_RDWR)
944
stderrcopy = os.dup(sys.stderr.fileno())
945
os.dup2(null, sys.stderr.fileno())
951
os.dup2(stderrcopy, sys.stderr.fileno())
954
def check_option_syntax(self, options):
955
check_option_syntax(self.parser, options)
957
def test_actions_all_conflicts_with_verbose(self):
958
for action, value in self.actions.items():
959
options = self.parser.parse_args()
960
setattr(options, action, value)
962
options.verbose = True
963
with self.assertParseError():
964
self.check_option_syntax(options)
966
def test_actions_with_client_conflicts_with_verbose(self):
967
for action, value in self.actions.items():
968
options = self.parser.parse_args()
969
setattr(options, action, value)
970
options.verbose = True
971
options.client = ["foo"]
972
with self.assertParseError():
973
self.check_option_syntax(options)
975
def test_dump_json_conflicts_with_verbose(self):
976
options = self.parser.parse_args()
977
options.dump_json = True
978
options.verbose = True
979
with self.assertParseError():
980
self.check_option_syntax(options)
982
def test_dump_json_conflicts_with_action(self):
983
for action, value in self.actions.items():
984
options = self.parser.parse_args()
985
setattr(options, action, value)
986
options.dump_json = True
987
with self.assertParseError():
988
self.check_option_syntax(options)
990
def test_all_can_not_be_alone(self):
991
options = self.parser.parse_args()
993
with self.assertParseError():
994
self.check_option_syntax(options)
996
def test_all_is_ok_with_any_action(self):
997
for action, value in self.actions.items():
998
options = self.parser.parse_args()
999
setattr(options, action, value)
1001
self.check_option_syntax(options)
1003
def test_any_action_is_ok_with_one_client(self):
1004
for action, value in self.actions.items():
1005
options = self.parser.parse_args()
1006
setattr(options, action, value)
1007
options.client = ["foo"]
1008
self.check_option_syntax(options)
1010
def test_actions_except_is_enabled_are_ok_with_two_clients(self):
1011
for action, value in self.actions.items():
1012
if action == "is_enabled":
1014
options = self.parser.parse_args()
1015
setattr(options, action, value)
1016
options.client = ["foo", "barbar"]
1017
self.check_option_syntax(options)
1019
def test_is_enabled_fails_without_client(self):
1020
options = self.parser.parse_args()
1021
options.is_enabled = True
1022
with self.assertParseError():
1023
self.check_option_syntax(options)
1025
def test_is_enabled_fails_with_two_clients(self):
1026
options = self.parser.parse_args()
1027
options.is_enabled = True
1028
options.client = ["foo", "barbar"]
1029
with self.assertParseError():
1030
self.check_option_syntax(options)
1032
def test_remove_can_only_be_combined_with_action_deny(self):
1033
for action, value in self.actions.items():
1034
if action in {"remove", "deny"}:
1036
options = self.parser.parse_args()
1037
setattr(options, action, value)
1039
options.remove = True
1040
with self.assertParseError():
1041
self.check_option_syntax(options)
1044
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1045
def test_calls_and_returns_get_object_on_bus(self):
1046
class MockBus(object):
1048
def get_object(mockbus_self, busname, dbus_path):
1049
# Note that "self" is still the testcase instance,
1050
# this MockBus instance is in "mockbus_self".
1051
self.assertEqual(busname, dbus_busname)
1052
self.assertEqual(dbus_path, server_dbus_path)
1053
mockbus_self.called = True
1056
mockbus = get_mandos_dbus_object(bus=MockBus())
1057
self.assertIsInstance(mockbus, MockBus)
1058
self.assertTrue(mockbus.called)
1060
def test_logs_and_exits_on_dbus_error(self):
1061
class MockBusFailing(object):
1062
def get_object(self, busname, dbus_path):
1063
raise dbus.exceptions.DBusException("Test")
1065
with self.assertLogs(log, logging.CRITICAL):
1066
with self.assertRaises(SystemExit) as e:
1067
bus = get_mandos_dbus_object(bus=MockBusFailing())
1069
if isinstance(e.exception.code, int):
1070
self.assertNotEqual(e.exception.code, 0)
1072
self.assertIsNotNone(e.exception.code)
1075
class Test_get_managed_objects(TestCaseWithAssertLogs):
1076
def test_calls_and_returns_GetManagedObjects(self):
1077
managed_objects = {"/clients/foo": { "Name": "foo"}}
1078
class MockObjectManager(object):
1079
def GetManagedObjects(self):
1080
return managed_objects
1081
retval = get_managed_objects(MockObjectManager())
1082
self.assertDictEqual(managed_objects, retval)
1084
def test_logs_and_exits_on_dbus_error(self):
1085
dbus_logger = logging.getLogger("dbus.proxies")
1087
class MockObjectManagerFailing(object):
1088
def GetManagedObjects(self):
1089
dbus_logger.error("Test")
1090
raise dbus.exceptions.DBusException("Test")
1092
class CountingHandler(logging.Handler):
1094
def emit(self, record):
1097
counting_handler = CountingHandler()
1099
dbus_logger.addHandler(counting_handler)
1102
with self.assertLogs(log, logging.CRITICAL) as watcher:
1103
with self.assertRaises(SystemExit) as e:
1104
get_managed_objects(MockObjectManagerFailing())
1106
dbus_logger.removeFilter(counting_handler)
1108
# Make sure the dbus logger was suppressed
1109
self.assertEqual(counting_handler.count, 0)
1111
# Test that the dbus_logger still works
1112
with self.assertLogs(dbus_logger, logging.ERROR):
1113
dbus_logger.error("Test")
1115
if isinstance(e.exception.code, int):
1116
self.assertNotEqual(e.exception.code, 0)
1118
self.assertIsNotNone(e.exception.code)
1121
class Test_commands_from_options(unittest.TestCase):
1123
self.parser = argparse.ArgumentParser()
1124
add_command_line_options(self.parser)
1126
def test_is_enabled(self):
1127
self.assert_command_from_args(["--is-enabled", "foo"],
1130
def assert_command_from_args(self, args, command_cls,
1132
"""Assert that parsing ARGS should result in an instance of
1133
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1134
options = self.parser.parse_args(args)
1135
check_option_syntax(self.parser, options)
1136
commands = commands_from_options(options)
1137
self.assertEqual(len(commands), 1)
1138
command = commands[0]
1139
self.assertIsInstance(command, command_cls)
1140
for key, value in cmd_attrs.items():
1141
self.assertEqual(getattr(command, key), value)
1143
def test_is_enabled_short(self):
1144
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1146
def test_approve(self):
1147
self.assert_command_from_args(["--approve", "foo"],
1150
def test_approve_short(self):
1151
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1153
def test_deny(self):
1154
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1156
def test_deny_short(self):
1157
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1159
def test_remove(self):
1160
self.assert_command_from_args(["--remove", "foo"],
1163
def test_deny_before_remove(self):
1164
options = self.parser.parse_args(["--deny", "--remove",
1166
check_option_syntax(self.parser, options)
1167
commands = commands_from_options(options)
1168
self.assertEqual(len(commands), 2)
1169
self.assertIsInstance(commands[0], DenyCmd)
1170
self.assertIsInstance(commands[1], RemoveCmd)
1172
def test_deny_before_remove_reversed(self):
1173
options = self.parser.parse_args(["--remove", "--deny",
1175
check_option_syntax(self.parser, options)
1176
commands = commands_from_options(options)
1177
self.assertEqual(len(commands), 2)
1178
self.assertIsInstance(commands[0], DenyCmd)
1179
self.assertIsInstance(commands[1], RemoveCmd)
1181
def test_remove_short(self):
1182
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1184
def test_dump_json(self):
1185
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1187
def test_enable(self):
1188
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1190
def test_enable_short(self):
1191
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1193
def test_disable(self):
1194
self.assert_command_from_args(["--disable", "foo"],
1197
def test_disable_short(self):
1198
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1200
def test_bump_timeout(self):
1201
self.assert_command_from_args(["--bump-timeout", "foo"],
1204
def test_bump_timeout_short(self):
1205
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1207
def test_start_checker(self):
1208
self.assert_command_from_args(["--start-checker", "foo"],
1211
def test_stop_checker(self):
1212
self.assert_command_from_args(["--stop-checker", "foo"],
1215
def test_approve_by_default(self):
1216
self.assert_command_from_args(["--approve-by-default", "foo"],
1217
ApproveByDefaultCmd)
1219
def test_deny_by_default(self):
1220
self.assert_command_from_args(["--deny-by-default", "foo"],
1223
def test_checker(self):
1224
self.assert_command_from_args(["--checker", ":", "foo"],
1225
SetCheckerCmd, value_to_set=":")
1227
def test_checker_empty(self):
1228
self.assert_command_from_args(["--checker", "", "foo"],
1229
SetCheckerCmd, value_to_set="")
1231
def test_checker_short(self):
1232
self.assert_command_from_args(["-c", ":", "foo"],
1233
SetCheckerCmd, value_to_set=":")
1235
def test_host(self):
1236
self.assert_command_from_args(["--host", "foo.example.org",
1238
value_to_set="foo.example.org")
1240
def test_host_short(self):
1241
self.assert_command_from_args(["-H", "foo.example.org",
1243
value_to_set="foo.example.org")
1245
def test_secret_devnull(self):
1246
self.assert_command_from_args(["--secret", os.path.devnull,
1247
"foo"], SetSecretCmd,
1250
def test_secret_tempfile(self):
1251
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1252
value = b"secret\0xyzzy\nbar"
1255
self.assert_command_from_args(["--secret", f.name,
1256
"foo"], SetSecretCmd,
1259
def test_secret_devnull_short(self):
1260
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1261
SetSecretCmd, value_to_set=b"")
1263
def test_secret_tempfile_short(self):
1264
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1265
value = b"secret\0xyzzy\nbar"
1268
self.assert_command_from_args(["-s", f.name, "foo"],
1272
def test_timeout(self):
1273
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1275
value_to_set=300000)
1277
def test_timeout_short(self):
1278
self.assert_command_from_args(["-t", "PT5M", "foo"],
1280
value_to_set=300000)
1282
def test_extended_timeout(self):
1283
self.assert_command_from_args(["--extended-timeout", "PT15M",
1285
SetExtendedTimeoutCmd,
1286
value_to_set=900000)
1288
def test_interval(self):
1289
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1291
value_to_set=120000)
1293
def test_interval_short(self):
1294
self.assert_command_from_args(["-i", "PT2M", "foo"],
1296
value_to_set=120000)
1298
def test_approval_delay(self):
1299
self.assert_command_from_args(["--approval-delay", "PT30S",
1300
"foo"], SetApprovalDelayCmd,
1303
def test_approval_duration(self):
1304
self.assert_command_from_args(["--approval-duration", "PT1S",
1305
"foo"], SetApprovalDurationCmd,
1308
def test_print_table(self):
1309
self.assert_command_from_args([], PrintTableCmd,
1312
def test_print_table_verbose(self):
1313
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1316
def test_print_table_verbose_short(self):
1317
self.assert_command_from_args(["-v"], PrintTableCmd,
790
1321
class TestCmd(unittest.TestCase):
791
1322
"""Abstract class for tests of command classes"""
792
1324
def setUp(self):
794
1326
class MockClient(object):
795
1327
def __init__(self, name, **attributes):
796
self.__dbus_object_path__ = "objpath_{}".format(name)
1328
self.__dbus_object_path__ = "/clients/{}".format(name)
797
1329
self.attributes = attributes
798
1330
self.attributes["Name"] = name
800
def Set(self, interface, property, value, dbus_interface):
801
testcase.assertEqual(interface, client_interface)
802
testcase.assertEqual(dbus_interface,
803
dbus.PROPERTIES_IFACE)
804
self.attributes[property] = value
805
def Get(self, interface, property, dbus_interface):
806
testcase.assertEqual(interface, client_interface)
807
testcase.assertEqual(dbus_interface,
808
dbus.PROPERTIES_IFACE)
809
return self.attributes[property]
1332
def Set(self, interface, propname, value, dbus_interface):
1333
testcase.assertEqual(interface, client_dbus_interface)
1334
testcase.assertEqual(dbus_interface,
1335
dbus.PROPERTIES_IFACE)
1336
self.attributes[propname] = value
1337
def Get(self, interface, propname, dbus_interface):
1338
testcase.assertEqual(interface, client_dbus_interface)
1339
testcase.assertEqual(dbus_interface,
1340
dbus.PROPERTIES_IFACE)
1341
return self.attributes[propname]
810
1342
def Approve(self, approve, dbus_interface):
811
testcase.assertEqual(dbus_interface, client_interface)
1343
testcase.assertEqual(dbus_interface,
1344
client_dbus_interface)
812
1345
self.calls.append(("Approve", (approve,
813
1346
dbus_interface)))
814
1347
self.client = MockClient(
854
1387
ApprovedByDefault=dbus.Boolean(False),
855
1388
LastApprovalRequest="2019-01-03T00:00:00",
856
1389
ApprovalDelay=30000,
857
ApprovalDuration=1000,
1390
ApprovalDuration=93785000,
859
1392
ExtendedTimeout=900000,
860
1393
Expires="2019-02-05T00:00:00",
861
1394
LastCheckerStatus=-2)
862
1395
self.clients = collections.OrderedDict(
864
(self.client, self.client.attributes),
865
(self.other_client, self.other_client.attributes),
1397
("/clients/foo", self.client.attributes),
1398
("/clients/barbar", self.other_client.attributes),
867
self.one_client = {self.client: self.client.attributes}
869
class TestPrintTableCmd(TestCmd):
870
def test_normal(self):
871
output = PrintTableCmd().output(self.clients)
872
expected_output = """
873
Name Enabled Timeout Last Successful Check
874
foo Yes 00:05:00 2019-02-03T00:00:00
875
barbar Yes 00:05:00 2019-02-04T00:00:00
877
self.assertEqual(output, expected_output)
878
def test_verbose(self):
879
output = PrintTableCmd(verbose=True).output(self.clients)
880
expected_output = """
881
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
882
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
883
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
885
self.assertEqual(output, expected_output)
886
def test_one_client(self):
887
output = PrintTableCmd().output(self.one_client)
888
expected_output = """
889
Name Enabled Timeout Last Successful Check
890
foo Yes 00:05:00 2019-02-03T00:00:00
892
self.assertEqual(output, expected_output)
1400
self.one_client = {"/clients/foo": self.client.attributes}
1406
def get_object(client_bus_name, path):
1407
self.assertEqual(client_bus_name, dbus_busname)
1409
# Note: "self" here is the TestCmd instance, not
1410
# the Bus instance, since this is a static method!
1411
"/clients/foo": self.client,
1412
"/clients/barbar": self.other_client,
1417
class TestIsEnabledCmd(TestCmd):
1418
def test_is_enabled(self):
1419
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1421
for client, properties
1422
in self.clients.items()))
1424
def test_is_enabled_run_exits_successfully(self):
1425
with self.assertRaises(SystemExit) as e:
1426
IsEnabledCmd().run(self.one_client)
1427
if e.exception.code is not None:
1428
self.assertEqual(e.exception.code, 0)
1430
self.assertIsNone(e.exception.code)
1432
def test_is_enabled_run_exits_with_failure(self):
1433
self.client.attributes["Enabled"] = dbus.Boolean(False)
1434
with self.assertRaises(SystemExit) as e:
1435
IsEnabledCmd().run(self.one_client)
1436
if isinstance(e.exception.code, int):
1437
self.assertNotEqual(e.exception.code, 0)
1439
self.assertIsNotNone(e.exception.code)
1442
class TestApproveCmd(TestCmd):
1443
def test_approve(self):
1444
ApproveCmd().run(self.clients, self.bus)
1445
for clientpath in self.clients:
1446
client = self.bus.get_object(dbus_busname, clientpath)
1447
self.assertIn(("Approve", (True, client_dbus_interface)),
1451
class TestDenyCmd(TestCmd):
1452
def test_deny(self):
1453
DenyCmd().run(self.clients, self.bus)
1454
for clientpath in self.clients:
1455
client = self.bus.get_object(dbus_busname, clientpath)
1456
self.assertIn(("Approve", (False, client_dbus_interface)),
1460
class TestRemoveCmd(TestCmd):
1461
def test_remove(self):
1462
class MockMandos(object):
1465
def RemoveClient(self, dbus_path):
1466
self.calls.append(("RemoveClient", (dbus_path,)))
1467
mandos = MockMandos()
1468
super(TestRemoveCmd, self).setUp()
1469
RemoveCmd().run(self.clients, self.bus, mandos)
1470
self.assertEqual(len(mandos.calls), 2)
1471
for clientpath in self.clients:
1472
self.assertIn(("RemoveClient", (clientpath,)),
894
1476
class TestDumpJSONCmd(TestCmd):
895
1477
def setUp(self):
946
1528
return super(TestDumpJSONCmd, self).setUp()
947
1530
def test_normal(self):
948
json_data = json.loads(DumpJSONCmd().output(self.clients))
1531
output = DumpJSONCmd().output(self.clients.values())
1532
json_data = json.loads(output)
949
1533
self.assertDictEqual(json_data, self.expected_json)
950
1535
def test_one_client(self):
951
clients = self.one_client
952
json_data = json.loads(DumpJSONCmd().output(clients))
1536
output = DumpJSONCmd().output(self.one_client.values())
1537
json_data = json.loads(output)
953
1538
expected_json = {"foo": self.expected_json["foo"]}
954
1539
self.assertDictEqual(json_data, expected_json)
956
class TestIsEnabledCmd(TestCmd):
957
def test_is_enabled(self):
958
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
959
for client, properties in self.clients.items()))
960
def test_is_enabled_run_exits_successfully(self):
961
with self.assertRaises(SystemExit) as e:
962
IsEnabledCmd().run(None, self.one_client)
963
if e.exception.code is not None:
964
self.assertEqual(e.exception.code, 0)
966
self.assertIsNone(e.exception.code)
967
def test_is_enabled_run_exits_with_failure(self):
968
self.client.attributes["Enabled"] = dbus.Boolean(False)
969
with self.assertRaises(SystemExit) as e:
970
IsEnabledCmd().run(None, self.one_client)
971
if isinstance(e.exception.code, int):
972
self.assertNotEqual(e.exception.code, 0)
974
self.assertIsNotNone(e.exception.code)
976
class TestRemoveCmd(TestCmd):
977
def test_remove(self):
978
class MockMandos(object):
981
def RemoveClient(self, dbus_path):
982
self.calls.append(("RemoveClient", (dbus_path,)))
983
mandos = MockMandos()
984
super(TestRemoveCmd, self).setUp()
985
RemoveCmd().run(mandos, self.clients)
986
self.assertEqual(len(mandos.calls), 2)
987
for client in self.clients:
988
self.assertIn(("RemoveClient",
989
(client.__dbus_object_path__,)),
992
class TestApproveCmd(TestCmd):
993
def test_approve(self):
994
ApproveCmd().run(None, self.clients)
995
for client in self.clients:
996
self.assertIn(("Approve", (True, client_interface)),
999
class TestDenyCmd(TestCmd):
1000
def test_deny(self):
1001
DenyCmd().run(None, self.clients)
1002
for client in self.clients:
1003
self.assertIn(("Approve", (False, client_interface)),
1006
class TestEnableCmd(TestCmd):
1007
def test_enable(self):
1008
for client in self.clients:
1009
client.attributes["Enabled"] = False
1011
EnableCmd().run(None, self.clients)
1013
for client in self.clients:
1014
self.assertTrue(client.attributes["Enabled"])
1016
class TestDisableCmd(TestCmd):
1017
def test_disable(self):
1018
DisableCmd().run(None, self.clients)
1020
for client in self.clients:
1021
self.assertFalse(client.attributes["Enabled"])
1023
class Unique(object):
1024
"""Class for objects which exist only to be unique objects, since
1025
unittest.mock.sentinel only exists in Python 3.3"""
1542
class TestPrintTableCmd(TestCmd):
1543
def test_normal(self):
1544
output = PrintTableCmd().output(self.clients.values())
1545
expected_output = "\n".join((
1546
"Name Enabled Timeout Last Successful Check",
1547
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1548
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1550
self.assertEqual(output, expected_output)
1552
def test_verbose(self):
1553
output = PrintTableCmd(verbose=True).output(
1554
self.clients.values())
1569
"Last Successful Check ",
1570
"2019-02-03T00:00:00 ",
1571
"2019-02-04T00:00:00 ",
1574
"2019-01-02T00:00:00 ",
1575
"2019-01-03T00:00:00 ",
1587
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1589
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1593
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1594
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1596
"Check Is Running ",
1601
"2019-01-03T00:00:00 ",
1602
"2019-01-04T00:00:00 ",
1604
"Approval Is Pending ",
1608
"Approved By Default ",
1612
"Last Approval Request ",
1614
"2019-01-03T00:00:00 ",
1620
"Approval Duration ",
1625
"fping -q -- %(host)s ",
1628
"Extended Timeout ",
1633
"2019-02-04T00:00:00 ",
1634
"2019-02-05T00:00:00 ",
1636
"Last Checker Status",
1641
num_lines = max(len(rows) for rows in columns)
1642
expected_output = "\n".join("".join(rows[line]
1643
for rows in columns)
1644
for line in range(num_lines))
1645
self.assertEqual(output, expected_output)
1647
def test_one_client(self):
1648
output = PrintTableCmd().output(self.one_client.values())
1649
expected_output = "\n".join((
1650
"Name Enabled Timeout Last Successful Check",
1651
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1653
self.assertEqual(output, expected_output)
1027
1656
class TestPropertyCmd(TestCmd):
1028
1657
"""Abstract class for tests of PropertyCmd classes"""
1033
1662
self.values_to_set)
1034
1663
for value_to_set, value_to_get in zip(self.values_to_set,
1035
1664
values_to_get):
1036
for client in self.clients:
1037
old_value = client.attributes[self.property]
1038
self.assertNotIsInstance(old_value, Unique)
1039
client.attributes[self.property] = Unique()
1665
for clientpath in self.clients:
1666
client = self.bus.get_object(dbus_busname, clientpath)
1667
old_value = client.attributes[self.propname]
1668
self.assertNotIsInstance(old_value, self.Unique)
1669
client.attributes[self.propname] = self.Unique()
1040
1670
self.run_command(value_to_set, self.clients)
1041
for client in self.clients:
1042
value = client.attributes[self.property]
1043
self.assertNotIsInstance(value, Unique)
1671
for clientpath in self.clients:
1672
client = self.bus.get_object(dbus_busname, clientpath)
1673
value = client.attributes[self.propname]
1674
self.assertNotIsInstance(value, self.Unique)
1044
1675
self.assertEqual(value, value_to_get)
1677
class Unique(object):
1678
"""Class for objects which exist only to be unique objects,
1679
since unittest.mock.sentinel only exists in Python 3.3"""
1045
1681
def run_command(self, value, clients):
1046
self.command().run(None, clients)
1682
self.command().run(clients, self.bus)
1685
class TestEnableCmd(TestPropertyCmd):
1687
propname = "Enabled"
1688
values_to_set = [dbus.Boolean(True)]
1691
class TestDisableCmd(TestPropertyCmd):
1692
command = DisableCmd
1693
propname = "Enabled"
1694
values_to_set = [dbus.Boolean(False)]
1048
1697
class TestBumpTimeoutCmd(TestPropertyCmd):
1049
1698
command = BumpTimeoutCmd
1050
property = "LastCheckedOK"
1699
propname = "LastCheckedOK"
1051
1700
values_to_set = [""]
1053
1703
class TestStartCheckerCmd(TestPropertyCmd):
1054
1704
command = StartCheckerCmd
1055
property = "CheckerRunning"
1705
propname = "CheckerRunning"
1056
1706
values_to_set = [dbus.Boolean(True)]
1058
1709
class TestStopCheckerCmd(TestPropertyCmd):
1059
1710
command = StopCheckerCmd
1060
property = "CheckerRunning"
1711
propname = "CheckerRunning"
1061
1712
values_to_set = [dbus.Boolean(False)]
1063
1715
class TestApproveByDefaultCmd(TestPropertyCmd):
1064
1716
command = ApproveByDefaultCmd
1065
property = "ApprovedByDefault"
1717
propname = "ApprovedByDefault"
1066
1718
values_to_set = [dbus.Boolean(True)]
1068
1721
class TestDenyByDefaultCmd(TestPropertyCmd):
1069
1722
command = DenyByDefaultCmd
1070
property = "ApprovedByDefault"
1723
propname = "ApprovedByDefault"
1071
1724
values_to_set = [dbus.Boolean(False)]
1073
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1074
"""Abstract class for tests of PropertyCmd classes using the
1075
ValueArgumentMixIn"""
1727
class TestPropertyValueCmd(TestPropertyCmd):
1728
"""Abstract class for tests of PropertyValueCmd classes"""
1076
1730
def runTest(self):
1077
if type(self) is TestValueArgumentPropertyCmd:
1731
if type(self) is TestPropertyValueCmd:
1079
return super(TestValueArgumentPropertyCmd, self).runTest()
1733
return super(TestPropertyValueCmd, self).runTest()
1080
1735
def run_command(self, value, clients):
1081
self.command(value).run(None, clients)
1083
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1736
self.command(value).run(clients, self.bus)
1739
class TestSetCheckerCmd(TestPropertyValueCmd):
1084
1740
command = SetCheckerCmd
1085
property = "Checker"
1741
propname = "Checker"
1086
1742
values_to_set = ["", ":", "fping -q -- %s"]
1088
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1745
class TestSetHostCmd(TestPropertyValueCmd):
1089
1746
command = SetHostCmd
1091
1748
values_to_set = ["192.0.2.3", "foo.example.org"]
1093
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1751
class TestSetSecretCmd(TestPropertyValueCmd):
1094
1752
command = SetSecretCmd
1096
values_to_set = [open("/dev/null", "rb"),
1754
values_to_set = [io.BytesIO(b""),
1097
1755
io.BytesIO(b"secret\0xyzzy\nbar")]
1098
1756
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1100
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1759
class TestSetTimeoutCmd(TestPropertyValueCmd):
1101
1760
command = SetTimeoutCmd
1102
property = "Timeout"
1103
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1106
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1761
propname = "Timeout"
1762
values_to_set = [datetime.timedelta(),
1763
datetime.timedelta(minutes=5),
1764
datetime.timedelta(seconds=1),
1765
datetime.timedelta(weeks=1),
1766
datetime.timedelta(weeks=52)]
1767
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1770
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1107
1771
command = SetExtendedTimeoutCmd
1108
property = "ExtendedTimeout"
1109
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1112
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1772
propname = "ExtendedTimeout"
1773
values_to_set = [datetime.timedelta(),
1774
datetime.timedelta(minutes=5),
1775
datetime.timedelta(seconds=1),
1776
datetime.timedelta(weeks=1),
1777
datetime.timedelta(weeks=52)]
1778
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1781
class TestSetIntervalCmd(TestPropertyValueCmd):
1113
1782
command = SetIntervalCmd
1114
property = "Interval"
1115
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1118
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1783
propname = "Interval"
1784
values_to_set = [datetime.timedelta(),
1785
datetime.timedelta(minutes=5),
1786
datetime.timedelta(seconds=1),
1787
datetime.timedelta(weeks=1),
1788
datetime.timedelta(weeks=52)]
1789
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1792
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1119
1793
command = SetApprovalDelayCmd
1120
property = "ApprovalDelay"
1121
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1122
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1124
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1794
propname = "ApprovalDelay"
1795
values_to_set = [datetime.timedelta(),
1796
datetime.timedelta(minutes=5),
1797
datetime.timedelta(seconds=1),
1798
datetime.timedelta(weeks=1),
1799
datetime.timedelta(weeks=52)]
1800
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1803
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1125
1804
command = SetApprovalDurationCmd
1126
property = "ApprovalDuration"
1127
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1128
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1130
class Test_command_from_options(unittest.TestCase):
1132
self.parser = argparse.ArgumentParser()
1133
add_command_line_options(self.parser)
1134
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1135
"""Assert that parsing ARGS should result in an instance of
1136
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1137
options = self.parser.parse_args(args)
1138
commands = commands_from_options(options)
1139
self.assertEqual(len(commands), 1)
1140
command = commands[0]
1141
self.assertIsInstance(command, command_cls)
1142
for key, value in cmd_attrs.items():
1143
self.assertEqual(getattr(command, key), value)
1144
def test_print_table(self):
1145
self.assert_command_from_args([], PrintTableCmd,
1148
def test_print_table_verbose(self):
1149
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1152
def test_enable(self):
1153
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1155
def test_disable(self):
1156
self.assert_command_from_args(["--disable", "foo"],
1159
def test_bump_timeout(self):
1160
self.assert_command_from_args(["--bump-timeout", "foo"],
1163
def test_start_checker(self):
1164
self.assert_command_from_args(["--start-checker", "foo"],
1167
def test_stop_checker(self):
1168
self.assert_command_from_args(["--stop-checker", "foo"],
1171
def test_remove(self):
1172
self.assert_command_from_args(["--remove", "foo"],
1175
def test_checker(self):
1176
self.assert_command_from_args(["--checker", ":", "foo"],
1177
SetCheckerCmd, value_to_set=":")
1179
def test_timeout(self):
1180
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1182
value_to_set=300000)
1184
def test_extended_timeout(self):
1185
self.assert_command_from_args(["--extended-timeout", "PT15M",
1187
SetExtendedTimeoutCmd,
1188
value_to_set=900000)
1190
def test_interval(self):
1191
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1193
value_to_set=120000)
1195
def test_approve_by_default(self):
1196
self.assert_command_from_args(["--approve-by-default", "foo"],
1197
ApproveByDefaultCmd)
1199
def test_deny_by_default(self):
1200
self.assert_command_from_args(["--deny-by-default", "foo"],
1203
def test_approval_delay(self):
1204
self.assert_command_from_args(["--approval-delay", "PT30S",
1205
"foo"], SetApprovalDelayCmd,
1208
def test_approval_duration(self):
1209
self.assert_command_from_args(["--approval-duration", "PT1S",
1210
"foo"], SetApprovalDurationCmd,
1213
def test_host(self):
1214
self.assert_command_from_args(["--host", "foo.example.org",
1216
value_to_set="foo.example.org")
1218
def test_secret_devnull(self):
1219
self.assert_command_from_args(["--secret", os.path.devnull,
1220
"foo"], SetSecretCmd,
1223
def test_secret_tempfile(self):
1224
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1225
value = b"secret\0xyzzy\nbar"
1228
self.assert_command_from_args(["--secret", f.name,
1229
"foo"], SetSecretCmd,
1232
def test_approve(self):
1233
self.assert_command_from_args(["--approve", "foo"],
1236
def test_deny(self):
1237
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1239
def test_dump_json(self):
1240
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1242
def test_is_enabled(self):
1243
self.assert_command_from_args(["--is-enabled", "foo"],
1805
propname = "ApprovalDuration"
1806
values_to_set = [datetime.timedelta(),
1807
datetime.timedelta(minutes=5),
1808
datetime.timedelta(seconds=1),
1809
datetime.timedelta(weeks=1),
1810
datetime.timedelta(weeks=52)]
1811
values_to_get = [0, 300000, 1000, 604800000, 31449600000]