79
75
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
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)
97
bus = dbus.SystemBus()
98
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
99
dbus_busname, server_dbus_path)
100
mandos_dbus_objc = bus.get_object(dbus_busname,
102
except dbus.exceptions.DBusException:
103
log.critical("Could not connect to Mandos server")
106
mandos_serv = dbus.Interface(mandos_dbus_objc,
107
dbus_interface=server_dbus_interface)
108
mandos_serv_object_manager = dbus.Interface(
109
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
112
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
113
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
114
with SilenceLogger("dbus.proxies"):
115
all_clients = {path: ifs_and_props[client_dbus_interface]
116
for path, ifs_and_props in
117
mandos_serv_object_manager
118
.GetManagedObjects().items()
119
if client_dbus_interface in ifs_and_props}
120
except dbus.exceptions.DBusException as e:
121
log.critical("Failed to access Mandos server through D-Bus:"
125
# Compile dict of (clients: properties) to process
129
clients = all_clients
131
for name in clientnames:
132
for objpath, properties in all_clients.items():
133
if properties["Name"] == name:
134
clients[objpath] = properties
137
log.critical("Client not found on server: %r", name)
140
# Run all commands on clients
141
commands = commands_from_options(options)
142
for command in commands:
143
command.run(clients, bus, mandos_serv)
146
def add_command_line_options(parser):
147
parser.add_argument("--version", action="version",
148
version="%(prog)s {}".format(version),
149
help="show version number and exit")
150
parser.add_argument("-a", "--all", action="store_true",
151
help="Select all clients")
152
parser.add_argument("-v", "--verbose", action="store_true",
153
help="Print all fields")
154
parser.add_argument("-j", "--dump-json", action="store_true",
155
help="Dump client data in JSON format")
156
enable_disable = parser.add_mutually_exclusive_group()
157
enable_disable.add_argument("-e", "--enable", action="store_true",
158
help="Enable client")
159
enable_disable.add_argument("-d", "--disable",
161
help="disable client")
162
parser.add_argument("-b", "--bump-timeout", action="store_true",
163
help="Bump timeout for client")
164
start_stop_checker = parser.add_mutually_exclusive_group()
165
start_stop_checker.add_argument("--start-checker",
167
help="Start checker for client")
168
start_stop_checker.add_argument("--stop-checker",
170
help="Stop checker for client")
171
parser.add_argument("-V", "--is-enabled", action="store_true",
172
help="Check if client is enabled")
173
parser.add_argument("-r", "--remove", action="store_true",
174
help="Remove client")
175
parser.add_argument("-c", "--checker",
176
help="Set checker command for client")
177
parser.add_argument("-t", "--timeout", type=string_to_delta,
178
help="Set timeout for client")
179
parser.add_argument("--extended-timeout", type=string_to_delta,
180
help="Set extended timeout for client")
181
parser.add_argument("-i", "--interval", type=string_to_delta,
182
help="Set checker interval for client")
183
approve_deny_default = parser.add_mutually_exclusive_group()
184
approve_deny_default.add_argument(
185
"--approve-by-default", action="store_true",
186
default=None, dest="approved_by_default",
187
help="Set client to be approved by default")
188
approve_deny_default.add_argument(
189
"--deny-by-default", action="store_false",
190
dest="approved_by_default",
191
help="Set client to be denied by default")
192
parser.add_argument("--approval-delay", type=string_to_delta,
193
help="Set delay before client approve/deny")
194
parser.add_argument("--approval-duration", type=string_to_delta,
195
help="Set duration of one client approval")
196
parser.add_argument("-H", "--host", help="Set host for client")
197
parser.add_argument("-s", "--secret",
198
type=argparse.FileType(mode="rb"),
199
help="Set password blob (file) for client")
200
approve_deny = parser.add_mutually_exclusive_group()
201
approve_deny.add_argument(
202
"-A", "--approve", action="store_true",
203
help="Approve any current client request")
204
approve_deny.add_argument("-D", "--deny", action="store_true",
205
help="Deny any current client request")
206
parser.add_argument("--debug", action="store_true",
207
help="Debug mode (show D-Bus commands)")
208
parser.add_argument("--check", action="store_true",
209
help="Run self-test")
210
parser.add_argument("client", nargs="*", help="Client name")
213
def string_to_delta(interval):
214
"""Parse a string and return a datetime.timedelta"""
217
return rfc3339_duration_to_delta(interval)
218
except ValueError as e:
219
log.warning("%s - Parsing as pre-1.6.1 interval instead",
221
return parse_pre_1_6_1_interval(interval)
78
def milliseconds_to_string(ms):
79
td = datetime.timedelta(0, 0, 0, ms)
80
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
81
.format(days="{}T".format(td.days) if td.days else "",
82
hours=td.seconds // 3600,
83
minutes=(td.seconds % 3600) // 60,
84
seconds=td.seconds % 60))
224
87
def rfc3339_duration_to_delta(duration):
397
def check_option_syntax(parser, options):
398
"""Apply additional restrictions on options, not expressible in
401
def has_actions(options):
402
return any((options.enable,
404
options.bump_timeout,
405
options.start_checker,
406
options.stop_checker,
409
options.checker is not None,
410
options.timeout is not None,
411
options.extended_timeout is not None,
412
options.interval is not None,
413
options.approved_by_default is not None,
414
options.approval_delay is not None,
415
options.approval_duration is not None,
416
options.host is not None,
417
options.secret is not None,
421
if has_actions(options) and not (options.client or options.all):
422
parser.error("Options require clients names or --all.")
423
if options.verbose and has_actions(options):
424
parser.error("--verbose can only be used alone.")
425
if options.dump_json and (options.verbose
426
or has_actions(options)):
427
parser.error("--dump-json can only be used alone.")
428
if options.all and not has_actions(options):
429
parser.error("--all requires an action.")
430
if options.is_enabled and len(options.client) > 1:
431
parser.error("--is-enabled requires exactly one client")
433
options.remove = False
434
if has_actions(options) and not options.deny:
435
parser.error("--remove can only be combined with --deny")
436
options.remove = True
439
class SilenceLogger(object):
440
"Simple context manager to silence a particular logger"
441
def __init__(self, loggername):
442
self.logger = logging.getLogger(loggername)
445
self.logger.addFilter(self.nullfilter)
448
class NullFilter(logging.Filter):
449
def filter(self, record):
452
nullfilter = NullFilter()
454
def __exit__(self, exc_type, exc_val, exc_tb):
455
self.logger.removeFilter(self.nullfilter)
458
def commands_from_options(options):
462
if options.is_enabled:
463
commands.append(IsEnabledCmd())
466
commands.append(ApproveCmd())
469
commands.append(DenyCmd())
472
commands.append(RemoveCmd())
474
if options.dump_json:
475
commands.append(DumpJSONCmd())
478
commands.append(EnableCmd())
481
commands.append(DisableCmd())
483
if options.bump_timeout:
484
commands.append(BumpTimeoutCmd())
486
if options.start_checker:
487
commands.append(StartCheckerCmd())
489
if options.stop_checker:
490
commands.append(StopCheckerCmd())
492
if options.approved_by_default is not None:
493
if options.approved_by_default:
494
commands.append(ApproveByDefaultCmd())
496
commands.append(DenyByDefaultCmd())
498
if options.checker is not None:
499
commands.append(SetCheckerCmd(options.checker))
501
if options.host is not None:
502
commands.append(SetHostCmd(options.host))
504
if options.secret is not None:
505
commands.append(SetSecretCmd(options.secret))
507
if options.timeout is not None:
508
commands.append(SetTimeoutCmd(options.timeout))
510
if options.extended_timeout:
512
SetExtendedTimeoutCmd(options.extended_timeout))
514
if options.interval is not None:
515
commands.append(SetIntervalCmd(options.interval))
517
if options.approval_delay is not None:
518
commands.append(SetApprovalDelayCmd(options.approval_delay))
520
if options.approval_duration is not None:
522
SetApprovalDurationCmd(options.approval_duration))
524
# If no command option has been given, show table of clients,
525
# optionally verbosely
527
commands.append(PrintTableCmd(verbose=options.verbose))
271
## Classes for commands.
273
# Abstract classes first
532
274
class Command(object):
533
275
"""Abstract class for commands"""
534
def run(self, clients, bus=None, mandos=None):
276
def run(self, mandos, clients):
535
277
"""Normal commands should implement run_on_one_client(), but
536
278
commands which want to operate on all clients at the same time
537
279
can override this run() method instead."""
538
280
self.mandos = mandos
539
for clientpath, properties in clients.items():
540
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
541
dbus_busname, str(clientpath))
542
client = bus.get_object(dbus_busname, clientpath)
281
for client, properties in clients.items():
543
282
self.run_on_one_client(client, properties)
546
class IsEnabledCmd(Command):
547
def run(self, clients, bus=None, mandos=None):
548
client, properties = next(iter(clients.items()))
549
if self.is_enabled(client, properties):
552
def is_enabled(self, client, properties):
553
return properties["Enabled"]
556
class ApproveCmd(Command):
557
def run_on_one_client(self, client, properties):
558
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
559
client.__dbus_object_path__, client_dbus_interface)
560
client.Approve(dbus.Boolean(True),
561
dbus_interface=client_dbus_interface)
564
class DenyCmd(Command):
565
def run_on_one_client(self, client, properties):
566
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
567
client.__dbus_object_path__, client_dbus_interface)
568
client.Approve(dbus.Boolean(False),
569
dbus_interface=client_dbus_interface)
572
class RemoveCmd(Command):
573
def run_on_one_client(self, client, properties):
574
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
575
server_dbus_path, server_dbus_interface,
576
str(client.__dbus_object_path__))
577
self.mandos.RemoveClient(client.__dbus_object_path__)
580
class OutputCmd(Command):
581
"""Abstract class for commands outputting client details"""
284
class PrintCmd(Command):
285
"""Abstract class for commands printing client details"""
582
286
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
583
287
"Created", "Interval", "Host", "KeyID",
584
288
"Fingerprint", "CheckerRunning", "LastEnabled",
693
403
**{key: self.string_from_client(client, key)
694
404
for key in self.keywords})
697
def milliseconds_to_string(ms):
698
td = datetime.timedelta(0, 0, 0, ms)
699
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
700
.format(days="{}T".format(td.days)
702
hours=td.seconds // 3600,
703
minutes=(td.seconds % 3600) // 60,
704
seconds=td.seconds % 60))
707
class PropertyCmd(Command):
708
"""Abstract class for Actions for setting one client property"""
710
def run_on_one_client(self, client, properties):
711
"""Set the Client's D-Bus property"""
712
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
713
client.__dbus_object_path__,
714
dbus.PROPERTIES_IFACE, client_dbus_interface,
715
self.propname, self.value_to_set
716
if not isinstance(self.value_to_set, dbus.Boolean)
717
else bool(self.value_to_set))
718
client.Set(client_dbus_interface, self.propname,
720
dbus_interface=dbus.PROPERTIES_IFACE)
724
raise NotImplementedError()
408
class DumpJSONCmd(PrintCmd):
409
def output(self, clients):
410
data = {client["Name"]:
411
{key: self.dbus_boolean_to_bool(client[key])
412
for key in self.all_keywords}
413
for client in clients.values()}
414
return json.dumps(data, indent=4, separators=(',', ': '))
416
def dbus_boolean_to_bool(value):
417
if isinstance(value, dbus.Boolean):
421
class IsEnabledCmd(Command):
422
def run_on_one_client(self, client, properties):
423
if self.is_enabled(client, properties):
426
def is_enabled(self, client, properties):
427
return bool(properties["Enabled"])
429
class RemoveCmd(Command):
430
def run_on_one_client(self, client, properties):
431
self.mandos.RemoveClient(client.__dbus_object_path__)
433
class ApproveCmd(Command):
434
def run_on_one_client(self, client, properties):
435
client.Approve(dbus.Boolean(True),
436
dbus_interface=client_interface)
438
class DenyCmd(Command):
439
def run_on_one_client(self, client, properties):
440
client.Approve(dbus.Boolean(False),
441
dbus_interface=client_interface)
727
443
class EnableCmd(PropertyCmd):
729
445
value_to_set = dbus.Boolean(True)
732
447
class DisableCmd(PropertyCmd):
734
449
value_to_set = dbus.Boolean(False)
737
451
class BumpTimeoutCmd(PropertyCmd):
738
propname = "LastCheckedOK"
452
property = "LastCheckedOK"
739
453
value_to_set = ""
742
455
class StartCheckerCmd(PropertyCmd):
743
propname = "CheckerRunning"
456
property = "CheckerRunning"
744
457
value_to_set = dbus.Boolean(True)
747
459
class StopCheckerCmd(PropertyCmd):
748
propname = "CheckerRunning"
460
property = "CheckerRunning"
749
461
value_to_set = dbus.Boolean(False)
752
463
class ApproveByDefaultCmd(PropertyCmd):
753
propname = "ApprovedByDefault"
464
property = "ApprovedByDefault"
754
465
value_to_set = dbus.Boolean(True)
757
467
class DenyByDefaultCmd(PropertyCmd):
758
propname = "ApprovedByDefault"
468
property = "ApprovedByDefault"
759
469
value_to_set = dbus.Boolean(False)
762
class PropertyValueCmd(PropertyCmd):
763
"""Abstract class for PropertyCmd recieving a value as argument"""
764
def __init__(self, value):
765
self.value_to_set = value
768
class SetCheckerCmd(PropertyValueCmd):
772
class SetHostCmd(PropertyValueCmd):
776
class SetSecretCmd(PropertyValueCmd):
780
def value_to_set(self):
784
def value_to_set(self, value):
785
"""When setting, read data from supplied file object"""
786
self._vts = value.read()
790
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
791
"""Abstract class for PropertyValueCmd taking a value argument as
792
a datetime.timedelta() but should store it as milliseconds."""
795
def value_to_set(self):
799
def value_to_set(self, value):
800
"""When setting, convert value from a datetime.timedelta"""
801
self._vts = int(round(value.total_seconds() * 1000))
804
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
808
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
809
propname = "ExtendedTimeout"
812
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
813
propname = "Interval"
816
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
817
propname = "ApprovalDelay"
820
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
821
propname = "ApprovalDuration"
471
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
474
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
477
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
480
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
483
class SetExtendedTimeoutCmd(PropertyCmd,
484
MillisecondsValueArgumentMixIn):
485
property = "ExtendedTimeout"
487
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
488
property = "Interval"
490
class SetApprovalDelayCmd(PropertyCmd,
491
MillisecondsValueArgumentMixIn):
492
property = "ApprovalDelay"
494
class SetApprovalDurationCmd(PropertyCmd,
495
MillisecondsValueArgumentMixIn):
496
property = "ApprovalDuration"
498
def has_actions(options):
499
return any((options.enable,
501
options.bump_timeout,
502
options.start_checker,
503
options.stop_checker,
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,
518
def add_command_line_options(parser):
519
parser.add_argument("--version", action="version",
520
version="%(prog)s {}".format(version),
521
help="show version number and exit")
522
parser.add_argument("-a", "--all", action="store_true",
523
help="Select all clients")
524
parser.add_argument("-v", "--verbose", action="store_true",
525
help="Print all fields")
526
parser.add_argument("-j", "--dump-json", action="store_true",
527
help="Dump client data in JSON format")
528
enable_disable = parser.add_mutually_exclusive_group()
529
enable_disable.add_argument("-e", "--enable", action="store_true",
530
help="Enable client")
531
enable_disable.add_argument("-d", "--disable",
533
help="disable client")
534
parser.add_argument("-b", "--bump-timeout", action="store_true",
535
help="Bump timeout for client")
536
start_stop_checker = parser.add_mutually_exclusive_group()
537
start_stop_checker.add_argument("--start-checker",
539
help="Start checker for client")
540
start_stop_checker.add_argument("--stop-checker",
542
help="Stop checker for client")
543
parser.add_argument("-V", "--is-enabled", action="store_true",
544
help="Check if client is enabled")
545
parser.add_argument("-r", "--remove", action="store_true",
546
help="Remove client")
547
parser.add_argument("-c", "--checker",
548
help="Set checker command for client")
549
parser.add_argument("-t", "--timeout",
550
help="Set timeout for client")
551
parser.add_argument("--extended-timeout",
552
help="Set extended timeout for client")
553
parser.add_argument("-i", "--interval",
554
help="Set checker interval for client")
555
approve_deny_default = parser.add_mutually_exclusive_group()
556
approve_deny_default.add_argument(
557
"--approve-by-default", action="store_true",
558
default=None, dest="approved_by_default",
559
help="Set client to be approved by default")
560
approve_deny_default.add_argument(
561
"--deny-by-default", action="store_false",
562
dest="approved_by_default",
563
help="Set client to be denied by default")
564
parser.add_argument("--approval-delay",
565
help="Set delay before client approve/deny")
566
parser.add_argument("--approval-duration",
567
help="Set duration of one client approval")
568
parser.add_argument("-H", "--host", help="Set host for client")
569
parser.add_argument("-s", "--secret",
570
type=argparse.FileType(mode="rb"),
571
help="Set password blob (file) for client")
572
approve_deny = parser.add_mutually_exclusive_group()
573
approve_deny.add_argument(
574
"-A", "--approve", action="store_true",
575
help="Approve any current client request")
576
approve_deny.add_argument("-D", "--deny", action="store_true",
577
help="Deny any current client request")
578
parser.add_argument("--check", action="store_true",
579
help="Run self-test")
580
parser.add_argument("client", nargs="*", help="Client name")
583
def commands_from_options(options):
587
if options.dump_json:
588
commands.append(DumpJSONCmd())
591
commands.append(EnableCmd())
594
commands.append(DisableCmd())
596
if options.bump_timeout:
597
commands.append(BumpTimeoutCmd())
599
if options.start_checker:
600
commands.append(StartCheckerCmd())
602
if options.stop_checker:
603
commands.append(StopCheckerCmd())
605
if options.is_enabled:
606
commands.append(IsEnabledCmd())
609
commands.append(RemoveCmd())
611
if options.checker is not None:
612
commands.append(SetCheckerCmd())
614
if options.timeout is not None:
615
commands.append(SetTimeoutCmd(options.timeout))
617
if options.extended_timeout:
619
SetExtendedTimeoutCmd(options.extended_timeout))
621
if options.interval is not None:
622
command.append(SetIntervalCmd(options.interval))
624
if options.approved_by_default is not None:
625
if options.approved_by_default:
626
command.append(ApproveByDefaultCmd())
628
command.append(DenyByDefaultCmd())
630
if options.approval_delay is not None:
631
command.append(SetApprovalDelayCmd(options.approval_delay))
633
if options.approval_duration is not None:
635
SetApprovalDurationCmd(options.approval_duration))
637
if options.host is not None:
638
command.append(SetHostCmd(options.host))
640
if options.secret is not None:
641
command.append(SetSecretCmd(options.secret))
644
commands.append(ApproveCmd())
647
commands.append(DenyCmd())
649
# If no command option has been given, show table of clients,
650
# optionally verbosely
652
commands.append(PrintTableCmd(verbose=options.verbose))
658
parser = argparse.ArgumentParser()
660
add_command_line_options(parser)
662
options = parser.parse_args()
664
if has_actions(options) and not (options.client or options.all):
665
parser.error("Options require clients names or --all.")
666
if options.verbose and has_actions(options):
667
parser.error("--verbose can only be used alone.")
668
if options.dump_json and (options.verbose
669
or has_actions(options)):
670
parser.error("--dump-json can only be used alone.")
671
if options.all and not has_actions(options):
672
parser.error("--all requires an action.")
673
if options.is_enabled and len(options.client) > 1:
674
parser.error("--is-enabled requires exactly one client")
676
clientnames = options.client
679
bus = dbus.SystemBus()
680
mandos_dbus_objc = bus.get_object(busname, server_path)
681
except dbus.exceptions.DBusException:
682
log.critical("Could not connect to Mandos server")
685
mandos_serv = dbus.Interface(mandos_dbus_objc,
686
dbus_interface=server_interface)
687
mandos_serv_object_manager = dbus.Interface(
688
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
690
# Filter out log message from dbus module
691
dbus_logger = logging.getLogger("dbus.proxies")
692
class NullFilter(logging.Filter):
693
def filter(self, record):
695
dbus_filter = NullFilter()
697
dbus_logger.addFilter(dbus_filter)
698
mandos_clients = {path: ifs_and_props[client_interface]
699
for path, ifs_and_props in
700
mandos_serv_object_manager
701
.GetManagedObjects().items()
702
if client_interface in ifs_and_props}
703
except dbus.exceptions.DBusException as e:
704
log.critical("Failed to access Mandos server through D-Bus:"
708
# restore dbus logger
709
dbus_logger.removeFilter(dbus_filter)
711
# Compile dict of (clients: properties) to process
715
clients = {bus.get_object(busname, path): properties
716
for path, properties in mandos_clients.items()}
718
for name in clientnames:
719
for path, client in mandos_clients.items():
720
if client["Name"] == name:
721
client_objc = bus.get_object(busname, path)
722
clients[client_objc] = client
725
log.critical("Client not found on server: %r", name)
728
# Run all commands on clients
729
commands = commands_from_options(options)
730
for command in commands:
731
command.run(mandos_serv, clients)
734
class Test_milliseconds_to_string(unittest.TestCase):
736
self.assertEqual(milliseconds_to_string(93785000),
738
def test_no_days(self):
739
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
740
def test_all_zero(self):
741
self.assertEqual(milliseconds_to_string(0), "00:00:00")
742
def test_no_fractional_seconds(self):
743
self.assertEqual(milliseconds_to_string(400), "00:00:00")
744
self.assertEqual(milliseconds_to_string(900), "00:00:00")
745
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
825
747
class Test_string_to_delta(unittest.TestCase):
826
748
def test_handles_basic_rfc3339(self):
827
749
self.assertEqual(string_to_delta("PT0S"),
856
777
self.assertEqual(value, datetime.timedelta(0, 7200))
859
class Test_check_option_syntax(unittest.TestCase):
861
self.parser = argparse.ArgumentParser()
862
add_command_line_options(self.parser)
864
def test_actions_requires_client_or_all(self):
865
for action, value in self.actions.items():
866
options = self.parser.parse_args()
867
setattr(options, action, value)
868
with self.assertParseError():
869
self.check_option_syntax(options)
871
# This mostly corresponds to the definition from has_actions() in
872
# check_option_syntax()
874
# The actual values set here are not that important, but we do
875
# at least stick to the correct types, even though they are
879
"bump_timeout": True,
880
"start_checker": True,
881
"stop_checker": True,
885
"timeout": datetime.timedelta(),
886
"extended_timeout": datetime.timedelta(),
887
"interval": datetime.timedelta(),
888
"approved_by_default": True,
889
"approval_delay": datetime.timedelta(),
890
"approval_duration": datetime.timedelta(),
892
"secret": io.BytesIO(b"x"),
897
@contextlib.contextmanager
898
def assertParseError(self):
899
with self.assertRaises(SystemExit) as e:
900
with self.temporarily_suppress_stderr():
902
# Exit code from argparse is guaranteed to be "2". Reference:
903
# https://docs.python.org/3/library
904
# /argparse.html#exiting-methods
905
self.assertEqual(e.exception.code, 2)
908
@contextlib.contextmanager
909
def temporarily_suppress_stderr():
910
null = os.open(os.path.devnull, os.O_RDWR)
911
stderrcopy = os.dup(sys.stderr.fileno())
912
os.dup2(null, sys.stderr.fileno())
918
os.dup2(stderrcopy, sys.stderr.fileno())
921
def check_option_syntax(self, options):
922
check_option_syntax(self.parser, options)
924
def test_actions_conflicts_with_verbose(self):
925
for action, value in self.actions.items():
926
options = self.parser.parse_args()
927
setattr(options, action, value)
928
options.verbose = True
929
with self.assertParseError():
930
self.check_option_syntax(options)
932
def test_dump_json_conflicts_with_verbose(self):
933
options = self.parser.parse_args()
934
options.dump_json = True
935
options.verbose = True
936
with self.assertParseError():
937
self.check_option_syntax(options)
939
def test_dump_json_conflicts_with_action(self):
940
for action, value in self.actions.items():
941
options = self.parser.parse_args()
942
setattr(options, action, value)
943
options.dump_json = True
944
with self.assertParseError():
945
self.check_option_syntax(options)
947
def test_all_can_not_be_alone(self):
948
options = self.parser.parse_args()
950
with self.assertParseError():
951
self.check_option_syntax(options)
953
def test_all_is_ok_with_any_action(self):
954
for action, value in self.actions.items():
955
options = self.parser.parse_args()
956
setattr(options, action, value)
958
self.check_option_syntax(options)
960
def test_is_enabled_fails_without_client(self):
961
options = self.parser.parse_args()
962
options.is_enabled = True
963
with self.assertParseError():
964
self.check_option_syntax(options)
966
def test_is_enabled_works_with_one_client(self):
967
options = self.parser.parse_args()
968
options.is_enabled = True
969
options.client = ["foo"]
970
self.check_option_syntax(options)
972
def test_is_enabled_fails_with_two_clients(self):
973
options = self.parser.parse_args()
974
options.is_enabled = True
975
options.client = ["foo", "barbar"]
976
with self.assertParseError():
977
self.check_option_syntax(options)
979
def test_remove_can_only_be_combined_with_action_deny(self):
980
for action, value in self.actions.items():
981
if action in {"remove", "deny"}:
983
options = self.parser.parse_args()
984
setattr(options, action, value)
986
options.remove = True
987
with self.assertParseError():
988
self.check_option_syntax(options)
991
class Test_SilenceLogger(unittest.TestCase):
992
loggername = "mandos-ctl.Test_SilenceLogger"
993
log = logging.getLogger(loggername)
994
log.propagate = False
995
log.addHandler(logging.NullHandler())
998
self.counting_filter = self.CountingFilter()
1000
class CountingFilter(logging.Filter):
1001
"Count number of records"
1003
def filter(self, record):
1007
def test_should_filter_records_only_when_active(self):
1009
with SilenceLogger(self.loggername):
1010
self.log.addFilter(self.counting_filter)
1011
self.log.info("Filtered log message 1")
1012
self.log.info("Non-filtered message 2")
1013
self.log.info("Non-filtered message 3")
1015
self.log.removeFilter(self.counting_filter)
1016
self.assertEqual(self.counting_filter.count, 2)
1019
class Test_commands_from_options(unittest.TestCase):
1021
self.parser = argparse.ArgumentParser()
1022
add_command_line_options(self.parser)
1024
def test_is_enabled(self):
1025
self.assert_command_from_args(["--is-enabled", "foo"],
1028
def assert_command_from_args(self, args, command_cls,
1030
"""Assert that parsing ARGS should result in an instance of
1031
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1032
options = self.parser.parse_args(args)
1033
check_option_syntax(self.parser, options)
1034
commands = commands_from_options(options)
1035
self.assertEqual(len(commands), 1)
1036
command = commands[0]
1037
self.assertIsInstance(command, command_cls)
1038
for key, value in cmd_attrs.items():
1039
self.assertEqual(getattr(command, key), value)
1041
def test_is_enabled_short(self):
1042
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1044
def test_approve(self):
1045
self.assert_command_from_args(["--approve", "foo"],
1048
def test_approve_short(self):
1049
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1051
def test_deny(self):
1052
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1054
def test_deny_short(self):
1055
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1057
def test_remove(self):
1058
self.assert_command_from_args(["--remove", "foo"],
1061
def test_deny_before_remove(self):
1062
options = self.parser.parse_args(["--deny", "--remove",
1064
check_option_syntax(self.parser, options)
1065
commands = commands_from_options(options)
1066
self.assertEqual(len(commands), 2)
1067
self.assertIsInstance(commands[0], DenyCmd)
1068
self.assertIsInstance(commands[1], RemoveCmd)
1070
def test_deny_before_remove_reversed(self):
1071
options = self.parser.parse_args(["--remove", "--deny",
1073
check_option_syntax(self.parser, options)
1074
commands = commands_from_options(options)
1075
self.assertEqual(len(commands), 2)
1076
self.assertIsInstance(commands[0], DenyCmd)
1077
self.assertIsInstance(commands[1], RemoveCmd)
1079
def test_remove_short(self):
1080
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1082
def test_dump_json(self):
1083
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1085
def test_enable(self):
1086
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1088
def test_enable_short(self):
1089
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1091
def test_disable(self):
1092
self.assert_command_from_args(["--disable", "foo"],
1095
def test_disable_short(self):
1096
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1098
def test_bump_timeout(self):
1099
self.assert_command_from_args(["--bump-timeout", "foo"],
1102
def test_bump_timeout_short(self):
1103
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1105
def test_start_checker(self):
1106
self.assert_command_from_args(["--start-checker", "foo"],
1109
def test_stop_checker(self):
1110
self.assert_command_from_args(["--stop-checker", "foo"],
1113
def test_approve_by_default(self):
1114
self.assert_command_from_args(["--approve-by-default", "foo"],
1115
ApproveByDefaultCmd)
1117
def test_deny_by_default(self):
1118
self.assert_command_from_args(["--deny-by-default", "foo"],
1121
def test_checker(self):
1122
self.assert_command_from_args(["--checker", ":", "foo"],
1123
SetCheckerCmd, value_to_set=":")
1125
def test_checker_empty(self):
1126
self.assert_command_from_args(["--checker", "", "foo"],
1127
SetCheckerCmd, value_to_set="")
1129
def test_checker_short(self):
1130
self.assert_command_from_args(["-c", ":", "foo"],
1131
SetCheckerCmd, value_to_set=":")
1133
def test_host(self):
1134
self.assert_command_from_args(["--host", "foo.example.org",
1136
value_to_set="foo.example.org")
1138
def test_host_short(self):
1139
self.assert_command_from_args(["-H", "foo.example.org",
1141
value_to_set="foo.example.org")
1143
def test_secret_devnull(self):
1144
self.assert_command_from_args(["--secret", os.path.devnull,
1145
"foo"], SetSecretCmd,
1148
def test_secret_tempfile(self):
1149
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1150
value = b"secret\0xyzzy\nbar"
1153
self.assert_command_from_args(["--secret", f.name,
1154
"foo"], SetSecretCmd,
1157
def test_secret_devnull_short(self):
1158
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1159
SetSecretCmd, value_to_set=b"")
1161
def test_secret_tempfile_short(self):
1162
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1163
value = b"secret\0xyzzy\nbar"
1166
self.assert_command_from_args(["-s", f.name, "foo"],
1170
def test_timeout(self):
1171
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1173
value_to_set=300000)
1175
def test_timeout_short(self):
1176
self.assert_command_from_args(["-t", "PT5M", "foo"],
1178
value_to_set=300000)
1180
def test_extended_timeout(self):
1181
self.assert_command_from_args(["--extended-timeout", "PT15M",
1183
SetExtendedTimeoutCmd,
1184
value_to_set=900000)
1186
def test_interval(self):
1187
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1189
value_to_set=120000)
1191
def test_interval_short(self):
1192
self.assert_command_from_args(["-i", "PT2M", "foo"],
1194
value_to_set=120000)
1196
def test_approval_delay(self):
1197
self.assert_command_from_args(["--approval-delay", "PT30S",
1198
"foo"], SetApprovalDelayCmd,
1201
def test_approval_duration(self):
1202
self.assert_command_from_args(["--approval-duration", "PT1S",
1203
"foo"], SetApprovalDurationCmd,
1206
def test_print_table(self):
1207
self.assert_command_from_args([], PrintTableCmd,
1210
def test_print_table_verbose(self):
1211
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1214
def test_print_table_verbose_short(self):
1215
self.assert_command_from_args(["-v"], PrintTableCmd,
1219
780
class TestCmd(unittest.TestCase):
1220
781
"""Abstract class for tests of command classes"""
1222
782
def setUp(self):
1224
784
class MockClient(object):
1225
785
def __init__(self, name, **attributes):
1226
self.__dbus_object_path__ = "/clients/{}".format(name)
786
self.__dbus_object_path__ = "objpath_{}".format(name)
1227
787
self.attributes = attributes
1228
788
self.attributes["Name"] = name
1230
def Set(self, interface, propname, value, dbus_interface):
1231
testcase.assertEqual(interface, client_dbus_interface)
1232
testcase.assertEqual(dbus_interface,
1233
dbus.PROPERTIES_IFACE)
1234
self.attributes[propname] = value
1235
def Get(self, interface, propname, dbus_interface):
1236
testcase.assertEqual(interface, client_dbus_interface)
1237
testcase.assertEqual(dbus_interface,
1238
dbus.PROPERTIES_IFACE)
1239
return self.attributes[propname]
790
def Set(self, interface, property, value, dbus_interface):
791
testcase.assertEqual(interface, client_interface)
792
testcase.assertEqual(dbus_interface,
793
dbus.PROPERTIES_IFACE)
794
self.attributes[property] = value
795
def Get(self, interface, property, dbus_interface):
796
testcase.assertEqual(interface, client_interface)
797
testcase.assertEqual(dbus_interface,
798
dbus.PROPERTIES_IFACE)
799
return self.attributes[property]
1240
800
def Approve(self, approve, dbus_interface):
1241
testcase.assertEqual(dbus_interface,
1242
client_dbus_interface)
801
testcase.assertEqual(dbus_interface, client_interface)
1243
802
self.calls.append(("Approve", (approve,
1244
803
dbus_interface)))
1245
804
self.client = MockClient(
1285
844
ApprovedByDefault=dbus.Boolean(False),
1286
845
LastApprovalRequest="2019-01-03T00:00:00",
1287
846
ApprovalDelay=30000,
1288
ApprovalDuration=93785000,
847
ApprovalDuration=1000,
1290
849
ExtendedTimeout=900000,
1291
850
Expires="2019-02-05T00:00:00",
1292
851
LastCheckerStatus=-2)
1293
852
self.clients = collections.OrderedDict(
1295
("/clients/foo", self.client.attributes),
1296
("/clients/barbar", self.other_client.attributes),
854
(self.client, self.client.attributes),
855
(self.other_client, self.other_client.attributes),
1298
self.one_client = {"/clients/foo": self.client.attributes}
1304
def get_object(client_bus_name, path):
1305
self.assertEqual(client_bus_name, dbus_busname)
1307
# Note: "self" here is the TestCmd instance, not
1308
# the Bus instance, since this is a static method!
1309
"/clients/foo": self.client,
1310
"/clients/barbar": self.other_client,
1315
class TestIsEnabledCmd(TestCmd):
1316
def test_is_enabled(self):
1317
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1319
for client, properties
1320
in self.clients.items()))
1322
def test_is_enabled_run_exits_successfully(self):
1323
with self.assertRaises(SystemExit) as e:
1324
IsEnabledCmd().run(self.one_client)
1325
if e.exception.code is not None:
1326
self.assertEqual(e.exception.code, 0)
1328
self.assertIsNone(e.exception.code)
1330
def test_is_enabled_run_exits_with_failure(self):
1331
self.client.attributes["Enabled"] = dbus.Boolean(False)
1332
with self.assertRaises(SystemExit) as e:
1333
IsEnabledCmd().run(self.one_client)
1334
if isinstance(e.exception.code, int):
1335
self.assertNotEqual(e.exception.code, 0)
1337
self.assertIsNotNone(e.exception.code)
1340
class TestApproveCmd(TestCmd):
1341
def test_approve(self):
1342
ApproveCmd().run(self.clients, self.bus)
1343
for clientpath in self.clients:
1344
client = self.bus.get_object(dbus_busname, clientpath)
1345
self.assertIn(("Approve", (True, client_dbus_interface)),
1349
class TestDenyCmd(TestCmd):
1350
def test_deny(self):
1351
DenyCmd().run(self.clients, self.bus)
1352
for clientpath in self.clients:
1353
client = self.bus.get_object(dbus_busname, clientpath)
1354
self.assertIn(("Approve", (False, client_dbus_interface)),
1358
class TestRemoveCmd(TestCmd):
1359
def test_remove(self):
1360
class MockMandos(object):
1363
def RemoveClient(self, dbus_path):
1364
self.calls.append(("RemoveClient", (dbus_path,)))
1365
mandos = MockMandos()
1366
super(TestRemoveCmd, self).setUp()
1367
RemoveCmd().run(self.clients, self.bus, mandos)
1368
self.assertEqual(len(mandos.calls), 2)
1369
for clientpath in self.clients:
1370
self.assertIn(("RemoveClient", (clientpath,)),
857
self.one_client = {self.client: self.client.attributes}
859
class TestPrintTableCmd(TestCmd):
860
def test_normal(self):
861
output = PrintTableCmd().output(self.clients)
862
expected_output = """
863
Name Enabled Timeout Last Successful Check
864
foo Yes 00:05:00 2019-02-03T00:00:00
865
barbar Yes 00:05:00 2019-02-04T00:00:00
867
self.assertEqual(output, expected_output)
868
def test_verbose(self):
869
output = PrintTableCmd(verbose=True).output(self.clients)
870
expected_output = """
871
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
872
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
873
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
875
self.assertEqual(output, expected_output)
876
def test_one_client(self):
877
output = PrintTableCmd().output(self.one_client)
878
expected_output = """
879
Name Enabled Timeout Last Successful Check
880
foo Yes 00:05:00 2019-02-03T00:00:00
882
self.assertEqual(output, expected_output)
1374
884
class TestDumpJSONCmd(TestCmd):
1375
885
def setUp(self):
1426
936
return super(TestDumpJSONCmd, self).setUp()
1428
937
def test_normal(self):
1429
output = DumpJSONCmd().output(self.clients.values())
1430
json_data = json.loads(output)
938
json_data = json.loads(DumpJSONCmd().output(self.clients))
1431
939
self.assertDictEqual(json_data, self.expected_json)
1433
940
def test_one_client(self):
1434
output = DumpJSONCmd().output(self.one_client.values())
1435
json_data = json.loads(output)
941
clients = self.one_client
942
json_data = json.loads(DumpJSONCmd().output(clients))
1436
943
expected_json = {"foo": self.expected_json["foo"]}
1437
944
self.assertDictEqual(json_data, expected_json)
1440
class TestPrintTableCmd(TestCmd):
1441
def test_normal(self):
1442
output = PrintTableCmd().output(self.clients.values())
1443
expected_output = "\n".join((
1444
"Name Enabled Timeout Last Successful Check",
1445
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1446
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1448
self.assertEqual(output, expected_output)
1450
def test_verbose(self):
1451
output = PrintTableCmd(verbose=True).output(
1452
self.clients.values())
1467
"Last Successful Check ",
1468
"2019-02-03T00:00:00 ",
1469
"2019-02-04T00:00:00 ",
1472
"2019-01-02T00:00:00 ",
1473
"2019-01-03T00:00:00 ",
1485
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1487
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1491
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1492
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1494
"Check Is Running ",
1499
"2019-01-03T00:00:00 ",
1500
"2019-01-04T00:00:00 ",
1502
"Approval Is Pending ",
1506
"Approved By Default ",
1510
"Last Approval Request ",
1512
"2019-01-03T00:00:00 ",
1518
"Approval Duration ",
1523
"fping -q -- %(host)s ",
1526
"Extended Timeout ",
1531
"2019-02-04T00:00:00 ",
1532
"2019-02-05T00:00:00 ",
1534
"Last Checker Status",
1539
num_lines = max(len(rows) for rows in columns)
1540
expected_output = "\n".join("".join(rows[line]
1541
for rows in columns)
1542
for line in range(num_lines))
1543
self.assertEqual(output, expected_output)
1545
def test_one_client(self):
1546
output = PrintTableCmd().output(self.one_client.values())
1547
expected_output = "\n".join((
1548
"Name Enabled Timeout Last Successful Check",
1549
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1551
self.assertEqual(output, expected_output)
946
class TestIsEnabledCmd(TestCmd):
947
def test_is_enabled(self):
948
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
949
for client, properties in self.clients.items()))
950
def test_is_enabled_run_exits_successfully(self):
951
with self.assertRaises(SystemExit) as e:
952
IsEnabledCmd().run(None, self.one_client)
953
if e.exception.code is not None:
954
self.assertEqual(e.exception.code, 0)
956
self.assertIsNone(e.exception.code)
957
def test_is_enabled_run_exits_with_failure(self):
958
self.client.attributes["Enabled"] = dbus.Boolean(False)
959
with self.assertRaises(SystemExit) as e:
960
IsEnabledCmd().run(None, self.one_client)
961
if isinstance(e.exception.code, int):
962
self.assertNotEqual(e.exception.code, 0)
964
self.assertIsNotNone(e.exception.code)
966
class TestRemoveCmd(TestCmd):
967
def test_remove(self):
968
class MockMandos(object):
971
def RemoveClient(self, dbus_path):
972
self.calls.append(("RemoveClient", (dbus_path,)))
973
mandos = MockMandos()
974
super(TestRemoveCmd, self).setUp()
975
RemoveCmd().run(mandos, self.clients)
976
self.assertEqual(len(mandos.calls), 2)
977
for client in self.clients:
978
self.assertIn(("RemoveClient",
979
(client.__dbus_object_path__,)),
982
class TestApproveCmd(TestCmd):
983
def test_approve(self):
984
ApproveCmd().run(None, self.clients)
985
for client in self.clients:
986
self.assertIn(("Approve", (True, client_interface)),
989
class TestDenyCmd(TestCmd):
991
DenyCmd().run(None, self.clients)
992
for client in self.clients:
993
self.assertIn(("Approve", (False, client_interface)),
996
class TestEnableCmd(TestCmd):
997
def test_enable(self):
998
for client in self.clients:
999
client.attributes["Enabled"] = False
1001
EnableCmd().run(None, self.clients)
1003
for client in self.clients:
1004
self.assertTrue(client.attributes["Enabled"])
1006
class TestDisableCmd(TestCmd):
1007
def test_disable(self):
1008
DisableCmd().run(None, self.clients)
1010
for client in self.clients:
1011
self.assertFalse(client.attributes["Enabled"])
1013
class Unique(object):
1014
"""Class for objects which exist only to be unique objects, since
1015
unittest.mock.sentinel only exists in Python 3.3"""
1554
1017
class TestPropertyCmd(TestCmd):
1555
1018
"""Abstract class for tests of PropertyCmd classes"""
1560
1023
self.values_to_set)
1561
1024
for value_to_set, value_to_get in zip(self.values_to_set,
1562
1025
values_to_get):
1563
for clientpath in self.clients:
1564
client = self.bus.get_object(dbus_busname, clientpath)
1565
old_value = client.attributes[self.propname]
1566
self.assertNotIsInstance(old_value, self.Unique)
1567
client.attributes[self.propname] = self.Unique()
1026
for client in self.clients:
1027
old_value = client.attributes[self.property]
1028
self.assertNotIsInstance(old_value, Unique)
1029
client.attributes[self.property] = Unique()
1568
1030
self.run_command(value_to_set, self.clients)
1569
for clientpath in self.clients:
1570
client = self.bus.get_object(dbus_busname, clientpath)
1571
value = client.attributes[self.propname]
1572
self.assertNotIsInstance(value, self.Unique)
1031
for client in self.clients:
1032
value = client.attributes[self.property]
1033
self.assertNotIsInstance(value, Unique)
1573
1034
self.assertEqual(value, value_to_get)
1575
class Unique(object):
1576
"""Class for objects which exist only to be unique objects,
1577
since unittest.mock.sentinel only exists in Python 3.3"""
1579
1035
def run_command(self, value, clients):
1580
self.command().run(clients, self.bus)
1583
class TestEnableCmd(TestPropertyCmd):
1585
propname = "Enabled"
1586
values_to_set = [dbus.Boolean(True)]
1589
class TestDisableCmd(TestPropertyCmd):
1590
command = DisableCmd
1591
propname = "Enabled"
1592
values_to_set = [dbus.Boolean(False)]
1036
self.command().run(None, clients)
1595
1038
class TestBumpTimeoutCmd(TestPropertyCmd):
1596
1039
command = BumpTimeoutCmd
1597
propname = "LastCheckedOK"
1040
property = "LastCheckedOK"
1598
1041
values_to_set = [""]
1601
1043
class TestStartCheckerCmd(TestPropertyCmd):
1602
1044
command = StartCheckerCmd
1603
propname = "CheckerRunning"
1045
property = "CheckerRunning"
1604
1046
values_to_set = [dbus.Boolean(True)]
1607
1048
class TestStopCheckerCmd(TestPropertyCmd):
1608
1049
command = StopCheckerCmd
1609
propname = "CheckerRunning"
1050
property = "CheckerRunning"
1610
1051
values_to_set = [dbus.Boolean(False)]
1613
1053
class TestApproveByDefaultCmd(TestPropertyCmd):
1614
1054
command = ApproveByDefaultCmd
1615
propname = "ApprovedByDefault"
1055
property = "ApprovedByDefault"
1616
1056
values_to_set = [dbus.Boolean(True)]
1619
1058
class TestDenyByDefaultCmd(TestPropertyCmd):
1620
1059
command = DenyByDefaultCmd
1621
propname = "ApprovedByDefault"
1060
property = "ApprovedByDefault"
1622
1061
values_to_set = [dbus.Boolean(False)]
1625
class TestPropertyValueCmd(TestPropertyCmd):
1626
"""Abstract class for tests of PropertyValueCmd classes"""
1063
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1064
"""Abstract class for tests of PropertyCmd classes using the
1065
ValueArgumentMixIn"""
1628
1066
def runTest(self):
1629
if type(self) is TestPropertyValueCmd:
1067
if type(self) is TestValueArgumentPropertyCmd:
1631
return super(TestPropertyValueCmd, self).runTest()
1069
return super(TestValueArgumentPropertyCmd, self).runTest()
1633
1070
def run_command(self, value, clients):
1634
self.command(value).run(clients, self.bus)
1637
class TestSetCheckerCmd(TestPropertyValueCmd):
1071
self.command(value).run(None, clients)
1073
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1638
1074
command = SetCheckerCmd
1639
propname = "Checker"
1075
property = "Checker"
1640
1076
values_to_set = ["", ":", "fping -q -- %s"]
1643
class TestSetHostCmd(TestPropertyValueCmd):
1078
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1644
1079
command = SetHostCmd
1646
1081
values_to_set = ["192.0.2.3", "foo.example.org"]
1649
class TestSetSecretCmd(TestPropertyValueCmd):
1083
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1650
1084
command = SetSecretCmd
1652
values_to_set = [io.BytesIO(b""),
1653
io.BytesIO(b"secret\0xyzzy\nbar")]
1654
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1657
class TestSetTimeoutCmd(TestPropertyValueCmd):
1086
values_to_set = [b"", b"secret"]
1088
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1658
1089
command = SetTimeoutCmd
1659
propname = "Timeout"
1660
values_to_set = [datetime.timedelta(),
1661
datetime.timedelta(minutes=5),
1662
datetime.timedelta(seconds=1),
1663
datetime.timedelta(weeks=1),
1664
datetime.timedelta(weeks=52)]
1665
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1668
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1090
property = "Timeout"
1091
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1092
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1094
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1669
1095
command = SetExtendedTimeoutCmd
1670
propname = "ExtendedTimeout"
1671
values_to_set = [datetime.timedelta(),
1672
datetime.timedelta(minutes=5),
1673
datetime.timedelta(seconds=1),
1674
datetime.timedelta(weeks=1),
1675
datetime.timedelta(weeks=52)]
1676
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1679
class TestSetIntervalCmd(TestPropertyValueCmd):
1096
property = "ExtendedTimeout"
1097
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1098
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1100
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1680
1101
command = SetIntervalCmd
1681
propname = "Interval"
1682
values_to_set = [datetime.timedelta(),
1683
datetime.timedelta(minutes=5),
1684
datetime.timedelta(seconds=1),
1685
datetime.timedelta(weeks=1),
1686
datetime.timedelta(weeks=52)]
1687
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1690
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1102
property = "Interval"
1103
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1106
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1691
1107
command = SetApprovalDelayCmd
1692
propname = "ApprovalDelay"
1693
values_to_set = [datetime.timedelta(),
1694
datetime.timedelta(minutes=5),
1695
datetime.timedelta(seconds=1),
1696
datetime.timedelta(weeks=1),
1697
datetime.timedelta(weeks=52)]
1698
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1701
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1108
property = "ApprovalDelay"
1109
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1112
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1702
1113
command = SetApprovalDurationCmd
1703
propname = "ApprovalDuration"
1704
values_to_set = [datetime.timedelta(),
1705
datetime.timedelta(minutes=5),
1706
datetime.timedelta(seconds=1),
1707
datetime.timedelta(weeks=1),
1708
datetime.timedelta(weeks=52)]
1709
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1114
property = "ApprovalDuration"
1115
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1118
class TestOptions(unittest.TestCase):
1120
self.parser = argparse.ArgumentParser()
1121
add_command_line_options(self.parser)
1122
def commands_from_args(self, args):
1123
self.options = self.parser.parse_args(args)
1124
return commands_from_options(self.options)
1125
def test_default_is_show_table(self):
1126
commands = self.commands_from_args([])
1127
self.assertEqual(len(commands), 1)
1128
command = commands[0]
1129
self.assertIsInstance(command, PrintTableCmd)
1130
self.assertEqual(command.verbose, False)
1131
def test_show_table_verbose(self):
1132
commands = self.commands_from_args(["--verbose"])
1133
self.assertEqual(len(commands), 1)
1134
command = commands[0]
1135
self.assertIsInstance(command, PrintTableCmd)
1136
self.assertEqual(command.verbose, True)
1137
def test_enable(self):
1138
commands = self.commands_from_args(["--enable", "foo"])
1139
self.assertEqual(len(commands), 1)
1140
command = commands[0]
1141
self.assertIsInstance(command, EnableCmd)
1142
self.assertEqual(self.options.client, ["foo"])