To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 1068
- compare with another revision
- download diff
- download tarball
- view history from revision 1068
- Committer: Teddy Hogeborn
- Date: 2019-03-15 21:59:44 UTC
- Revision ID: teddy@recompile.se-20190315215944-lomtn36y7o3jphp8
mandos-ctl: Refactor
* mandos-ctl (main): Call get_mandos_dbus_object() to get
"mandos_dbus_objc", and rename it to
"mandos_dbus_object".
(get_mandos_dbus_object): New.
(Test_get_mandos_dbus_object): - '' -
* mandos-ctl (main): Call get_mandos_dbus_object() to get
"mandos_dbus_objc", and rename it to
"mandos_dbus_object".
(get_mandos_dbus_object): New.
(Test_get_mandos_dbus_object): - '' -
- files modified:
- mandos-ctl
added
removed
mandos-ctl
44
44
import logging
45
45
import io
46
46
import tempfile
47
import contextlib
47
48
48
49
import dbus
49
50
63
64
64
65
locale.setlocale(locale.LC_ALL, "")
65
66
66
domain = "se.recompile"
67
busname = domain + ".Mandos"
68
server_path = "/"
69
server_interface = domain + ".Mandos"
70
client_interface = domain + ".Mandos.Client"
67
dbus_busname_domain = "se.recompile"
68
dbus_busname = dbus_busname_domain + ".Mandos"
69
server_dbus_path = "/"
70
server_dbus_interface = dbus_busname_domain + ".Mandos"
71
client_dbus_interface = dbus_busname_domain + ".Mandos.Client"
72
del dbus_busname_domain
71
73
version = "1.8.3"
72
74
73
75
77
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
78
80
79
81
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))
82
def main():
83
parser = argparse.ArgumentParser()
84
85
add_command_line_options(parser)
86
87
options = parser.parse_args()
88
89
check_option_syntax(parser, options)
90
91
clientnames = options.client
92
93
if options.debug:
94
log.setLevel(logging.DEBUG)
95
96
bus = dbus.SystemBus()
97
98
mandos_dbus_object = get_mandos_dbus_object(bus)
99
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)
104
105
try:
106
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
107
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
108
with SilenceLogger("dbus.proxies"):
109
all_clients = {path: ifs_and_props[client_dbus_interface]
110
for path, ifs_and_props in
111
mandos_serv_object_manager
112
.GetManagedObjects().items()
113
if client_dbus_interface in ifs_and_props}
114
except dbus.exceptions.DBusException as e:
115
log.critical("Failed to access Mandos server through D-Bus:"
116
"\n%s", e)
117
sys.exit(1)
118
119
# Compile dict of (clients: properties) to process
120
clients = {}
121
122
if not clientnames:
123
clients = all_clients
124
else:
125
for name in clientnames:
126
for objpath, properties in all_clients.items():
127
if properties["Name"] == name:
128
clients[objpath] = properties
129
break
130
else:
131
log.critical("Client not found on server: %r", name)
132
sys.exit(1)
133
134
# Run all commands on clients
135
commands = commands_from_options(options)
136
for command in commands:
137
command.run(clients, bus, mandos_serv)
138
139
140
def add_command_line_options(parser):
141
parser.add_argument("--version", action="version",
142
version="%(prog)s {}".format(version),
143
help="show version number and exit")
144
parser.add_argument("-a", "--all", action="store_true",
145
help="Select all clients")
146
parser.add_argument("-v", "--verbose", action="store_true",
147
help="Print all fields")
148
parser.add_argument("-j", "--dump-json", action="store_true",
149
help="Dump client data in JSON format")
150
enable_disable = parser.add_mutually_exclusive_group()
151
enable_disable.add_argument("-e", "--enable", action="store_true",
152
help="Enable client")
153
enable_disable.add_argument("-d", "--disable",
154
action="store_true",
155
help="disable client")
156
parser.add_argument("-b", "--bump-timeout", action="store_true",
157
help="Bump timeout for client")
158
start_stop_checker = parser.add_mutually_exclusive_group()
159
start_stop_checker.add_argument("--start-checker",
160
action="store_true",
161
help="Start checker for client")
162
start_stop_checker.add_argument("--stop-checker",
163
action="store_true",
164
help="Stop checker for client")
165
parser.add_argument("-V", "--is-enabled", action="store_true",
166
help="Check if client is enabled")
167
parser.add_argument("-r", "--remove", action="store_true",
168
help="Remove client")
169
parser.add_argument("-c", "--checker",
170
help="Set checker command for client")
171
parser.add_argument("-t", "--timeout", type=string_to_delta,
172
help="Set timeout for client")
173
parser.add_argument("--extended-timeout", type=string_to_delta,
174
help="Set extended timeout for client")
175
parser.add_argument("-i", "--interval", type=string_to_delta,
176
help="Set checker interval for client")
177
approve_deny_default = parser.add_mutually_exclusive_group()
178
approve_deny_default.add_argument(
179
"--approve-by-default", action="store_true",
180
default=None, dest="approved_by_default",
181
help="Set client to be approved by default")
182
approve_deny_default.add_argument(
183
"--deny-by-default", action="store_false",
184
dest="approved_by_default",
185
help="Set client to be denied by default")
186
parser.add_argument("--approval-delay", type=string_to_delta,
187
help="Set delay before client approve/deny")
188
parser.add_argument("--approval-duration", type=string_to_delta,
189
help="Set duration of one client approval")
190
parser.add_argument("-H", "--host", help="Set host for client")
191
parser.add_argument("-s", "--secret",
192
type=argparse.FileType(mode="rb"),
193
help="Set password blob (file) for client")
194
approve_deny = parser.add_mutually_exclusive_group()
195
approve_deny.add_argument(
196
"-A", "--approve", action="store_true",
197
help="Approve any current client request")
198
approve_deny.add_argument("-D", "--deny", action="store_true",
199
help="Deny any current client request")
200
parser.add_argument("--debug", action="store_true",
201
help="Debug mode (show D-Bus commands)")
202
parser.add_argument("--check", action="store_true",
203
help="Run self-test")
204
parser.add_argument("client", nargs="*", help="Client name")
205
206
207
def string_to_delta(interval):
208
"""Parse a string and return a datetime.timedelta"""
209
210
try:
211
return rfc3339_duration_to_delta(interval)
212
except ValueError as e:
213
log.warning("%s - Parsing as pre-1.6.1 interval instead",
214
' '.join(e.args))
215
return parse_pre_1_6_1_interval(interval)
87
216
88
217
89
218
def rfc3339_duration_to_delta(duration):
216
345
return value
217
346
218
347
219
def string_to_delta(interval):
220
"""Parse a string and return a datetime.timedelta"""
221
222
try:
223
return rfc3339_duration_to_delta(interval)
224
except ValueError as e:
225
log.warning("%s - Parsing as pre-1.6.1 interval instead",
226
' '.join(e.args))
227
return parse_pre_1_6_1_interval(interval)
228
229
230
348
def parse_pre_1_6_1_interval(interval):
231
349
"""Parse an interval string as documented by Mandos before 1.6.1,
232
350
and return a datetime.timedelta
270
388
return value
271
389
272
390
273
## Classes for commands.
274
275
# Abstract classes first
391
def check_option_syntax(parser, options):
392
"""Apply additional restrictions on options, not expressible in
393
argparse"""
394
395
def has_actions(options):
396
return any((options.enable,
397
options.disable,
398
options.bump_timeout,
399
options.start_checker,
400
options.stop_checker,
401
options.is_enabled,
402
options.remove,
403
options.checker is not None,
404
options.timeout is not None,
405
options.extended_timeout is not None,
406
options.interval is not None,
407
options.approved_by_default is not None,
408
options.approval_delay is not None,
409
options.approval_duration is not None,
410
options.host is not None,
411
options.secret is not None,
412
options.approve,
413
options.deny))
414
415
if has_actions(options) and not (options.client or options.all):
416
parser.error("Options require clients names or --all.")
417
if options.verbose and has_actions(options):
418
parser.error("--verbose can only be used alone.")
419
if options.dump_json and (options.verbose
420
or has_actions(options)):
421
parser.error("--dump-json can only be used alone.")
422
if options.all and not has_actions(options):
423
parser.error("--all requires an action.")
424
if options.is_enabled and len(options.client) > 1:
425
parser.error("--is-enabled requires exactly one client")
426
if options.remove:
427
options.remove = False
428
if has_actions(options) and not options.deny:
429
parser.error("--remove can only be combined with --deny")
430
options.remove = True
431
432
433
def get_mandos_dbus_object(bus):
434
try:
435
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
436
dbus_busname, server_dbus_path)
437
mandos_dbus_object = bus.get_object(dbus_busname,
438
server_dbus_path)
439
except dbus.exceptions.DBusException:
440
log.critical("Could not connect to Mandos server")
441
sys.exit(1)
442
443
return mandos_dbus_object
444
445
446
class SilenceLogger(object):
447
"Simple context manager to silence a particular logger"
448
def __init__(self, loggername):
449
self.logger = logging.getLogger(loggername)
450
451
def __enter__(self):
452
self.logger.addFilter(self.nullfilter)
453
return self
454
455
class NullFilter(logging.Filter):
456
def filter(self, record):
457
return False
458
459
nullfilter = NullFilter()
460
461
def __exit__(self, exc_type, exc_val, exc_tb):
462
self.logger.removeFilter(self.nullfilter)
463
464
465
def commands_from_options(options):
466
467
commands = []
468
469
if options.is_enabled:
470
commands.append(IsEnabledCmd())
471
472
if options.approve:
473
commands.append(ApproveCmd())
474
475
if options.deny:
476
commands.append(DenyCmd())
477
478
if options.remove:
479
commands.append(RemoveCmd())
480
481
if options.dump_json:
482
commands.append(DumpJSONCmd())
483
484
if options.enable:
485
commands.append(EnableCmd())
486
487
if options.disable:
488
commands.append(DisableCmd())
489
490
if options.bump_timeout:
491
commands.append(BumpTimeoutCmd())
492
493
if options.start_checker:
494
commands.append(StartCheckerCmd())
495
496
if options.stop_checker:
497
commands.append(StopCheckerCmd())
498
499
if options.approved_by_default is not None:
500
if options.approved_by_default:
501
commands.append(ApproveByDefaultCmd())
502
else:
503
commands.append(DenyByDefaultCmd())
504
505
if options.checker is not None:
506
commands.append(SetCheckerCmd(options.checker))
507
508
if options.host is not None:
509
commands.append(SetHostCmd(options.host))
510
511
if options.secret is not None:
512
commands.append(SetSecretCmd(options.secret))
513
514
if options.timeout is not None:
515
commands.append(SetTimeoutCmd(options.timeout))
516
517
if options.extended_timeout:
518
commands.append(
519
SetExtendedTimeoutCmd(options.extended_timeout))
520
521
if options.interval is not None:
522
commands.append(SetIntervalCmd(options.interval))
523
524
if options.approval_delay is not None:
525
commands.append(SetApprovalDelayCmd(options.approval_delay))
526
527
if options.approval_duration is not None:
528
commands.append(
529
SetApprovalDurationCmd(options.approval_duration))
530
531
# If no command option has been given, show table of clients,
532
# optionally verbosely
533
if not commands:
534
commands.append(PrintTableCmd(verbose=options.verbose))
535
536
return commands
537
538
276
539
class Command(object):
277
540
"""Abstract class for commands"""
278
def run(self, mandos, clients):
541
def run(self, clients, bus=None, mandos=None):
279
542
"""Normal commands should implement run_on_one_client(), but
280
543
commands which want to operate on all clients at the same time
281
544
can override this run() method instead."""
282
545
self.mandos = mandos
283
for client, properties in clients.items():
546
for clientpath, properties in clients.items():
547
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
548
dbus_busname, str(clientpath))
549
client = bus.get_object(dbus_busname, clientpath)
284
550
self.run_on_one_client(client, properties)
285
551
286
class PrintCmd(Command):
287
"""Abstract class for commands printing client details"""
552
553
class IsEnabledCmd(Command):
554
def run(self, clients, bus=None, mandos=None):
555
client, properties = next(iter(clients.items()))
556
if self.is_enabled(client, properties):
557
sys.exit(0)
558
sys.exit(1)
559
def is_enabled(self, client, properties):
560
return properties["Enabled"]
561
562
563
class ApproveCmd(Command):
564
def run_on_one_client(self, client, properties):
565
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
566
client.__dbus_object_path__, client_dbus_interface)
567
client.Approve(dbus.Boolean(True),
568
dbus_interface=client_dbus_interface)
569
570
571
class DenyCmd(Command):
572
def run_on_one_client(self, client, properties):
573
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
574
client.__dbus_object_path__, client_dbus_interface)
575
client.Approve(dbus.Boolean(False),
576
dbus_interface=client_dbus_interface)
577
578
579
class RemoveCmd(Command):
580
def run_on_one_client(self, client, properties):
581
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
582
server_dbus_path, server_dbus_interface,
583
str(client.__dbus_object_path__))
584
self.mandos.RemoveClient(client.__dbus_object_path__)
585
586
587
class OutputCmd(Command):
588
"""Abstract class for commands outputting client details"""
288
589
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
289
590
"Created", "Interval", "Host", "KeyID",
290
591
"Fingerprint", "CheckerRunning", "LastEnabled",
292
593
"LastApprovalRequest", "ApprovalDelay",
293
594
"ApprovalDuration", "Checker", "ExtendedTimeout",
294
595
"Expires", "LastCheckerStatus")
295
def run(self, mandos, clients):
296
print(self.output(clients))
297
298
class PropertyCmd(Command):
299
"""Abstract class for Actions for setting one client property"""
300
def run_on_one_client(self, client, properties):
301
"""Set the Client's D-Bus property"""
302
client.Set(client_interface, self.property, self.value_to_set,
303
dbus_interface=dbus.PROPERTIES_IFACE)
304
305
class ValueArgumentMixIn(object):
306
"""Mixin class for commands taking a value as argument"""
307
def __init__(self, value):
308
self.value_to_set = value
309
310
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
311
"""Mixin class for commands taking a value argument as
312
milliseconds."""
313
@property
314
def value_to_set(self):
315
return self._vts
316
@value_to_set.setter
317
def value_to_set(self, value):
318
"""When setting, convert value to a datetime.timedelta"""
319
self._vts = int(round(value.total_seconds() * 1000))
320
321
# Actual (non-abstract) command classes
322
323
class PrintTableCmd(PrintCmd):
596
597
def run(self, clients, bus=None, mandos=None):
598
print(self.output(clients.values()))
599
600
def output(self, clients):
601
raise NotImplementedError()
602
603
604
class DumpJSONCmd(OutputCmd):
605
def output(self, clients):
606
data = {client["Name"]:
607
{key: self.dbus_boolean_to_bool(client[key])
608
for key in self.all_keywords}
609
for client in clients}
610
return json.dumps(data, indent=4, separators=(',', ': '))
611
612
@staticmethod
613
def dbus_boolean_to_bool(value):
614
if isinstance(value, dbus.Boolean):
615
value = bool(value)
616
return value
617
618
619
class PrintTableCmd(OutputCmd):
324
620
def __init__(self, verbose=False):
325
621
self.verbose = verbose
326
622
327
623
def output(self, clients):
328
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
624
default_keywords = ("Name", "Enabled", "Timeout",
625
"LastCheckedOK")
329
626
keywords = default_keywords
330
627
if self.verbose:
331
628
keywords = self.all_keywords
332
return str(self.TableOfClients(clients.values(), keywords))
629
return str(self.TableOfClients(clients, keywords))
333
630
334
631
class TableOfClients(object):
335
632
tableheaders = {
355
652
"LastCheckerStatus": "Last Checker Status",
356
653
}
357
654
358
def __init__(self, clients, keywords, tableheaders=None):
655
def __init__(self, clients, keywords):
359
656
self.clients = clients
360
657
self.keywords = keywords
361
if tableheaders is not None:
362
self.tableheaders = tableheaders
363
658
364
659
def __str__(self):
365
660
return "\n".join(self.rows())
388
683
def string_from_client(self, client, key):
389
684
return self.valuetostring(client[key], key)
390
685
391
@staticmethod
392
def valuetostring(value, keyword):
686
@classmethod
687
def valuetostring(cls, value, keyword):
393
688
if isinstance(value, dbus.Boolean):
394
689
return "Yes" if value else "No"
395
690
if keyword in ("Timeout", "Interval", "ApprovalDelay",
396
691
"ApprovalDuration", "ExtendedTimeout"):
397
return milliseconds_to_string(value)
692
return cls.milliseconds_to_string(value)
398
693
return str(value)
399
694
400
695
def header_line(self, format_string):
405
700
**{key: self.string_from_client(client, key)
406
701
for key in self.keywords})
407
702
408
409
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=(',', ': '))
417
@staticmethod
418
def dbus_boolean_to_bool(value):
419
if isinstance(value, dbus.Boolean):
420
value = bool(value)
421
return value
422
423
class IsEnabledCmd(Command):
424
def run_on_one_client(self, client, properties):
425
if self.is_enabled(client, properties):
426
sys.exit(0)
427
sys.exit(1)
428
def is_enabled(self, client, properties):
429
return bool(properties["Enabled"])
430
431
class RemoveCmd(Command):
432
def run_on_one_client(self, client, properties):
433
self.mandos.RemoveClient(client.__dbus_object_path__)
434
435
class ApproveCmd(Command):
436
def run_on_one_client(self, client, properties):
437
client.Approve(dbus.Boolean(True),
438
dbus_interface=client_interface)
439
440
class DenyCmd(Command):
441
def run_on_one_client(self, client, properties):
442
client.Approve(dbus.Boolean(False),
443
dbus_interface=client_interface)
703
@staticmethod
704
def milliseconds_to_string(ms):
705
td = datetime.timedelta(0, 0, 0, ms)
706
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
707
.format(days="{}T".format(td.days)
708
if td.days else "",
709
hours=td.seconds // 3600,
710
minutes=(td.seconds % 3600) // 60,
711
seconds=td.seconds % 60))
712
713
714
class PropertyCmd(Command):
715
"""Abstract class for Actions for setting one client property"""
716
717
def run_on_one_client(self, client, properties):
718
"""Set the Client's D-Bus property"""
719
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
720
client.__dbus_object_path__,
721
dbus.PROPERTIES_IFACE, client_dbus_interface,
722
self.propname, self.value_to_set
723
if not isinstance(self.value_to_set, dbus.Boolean)
724
else bool(self.value_to_set))
725
client.Set(client_dbus_interface, self.propname,
726
self.value_to_set,
727
dbus_interface=dbus.PROPERTIES_IFACE)
728
729
@property
730
def propname(self):
731
raise NotImplementedError()
732
444
733
445
734
class EnableCmd(PropertyCmd):
446
property = "Enabled"
735
propname = "Enabled"
447
736
value_to_set = dbus.Boolean(True)
448
737
738
449
739
class DisableCmd(PropertyCmd):
450
property = "Enabled"
740
propname = "Enabled"
451
741
value_to_set = dbus.Boolean(False)
452
742
743
453
744
class BumpTimeoutCmd(PropertyCmd):
454
property = "LastCheckedOK"
745
propname = "LastCheckedOK"
455
746
value_to_set = ""
456
747
748
457
749
class StartCheckerCmd(PropertyCmd):
458
property = "CheckerRunning"
750
propname = "CheckerRunning"
459
751
value_to_set = dbus.Boolean(True)
460
752
753
461
754
class StopCheckerCmd(PropertyCmd):
462
property = "CheckerRunning"
755
propname = "CheckerRunning"
463
756
value_to_set = dbus.Boolean(False)
464
757
758
465
759
class ApproveByDefaultCmd(PropertyCmd):
466
property = "ApprovedByDefault"
760
propname = "ApprovedByDefault"
467
761
value_to_set = dbus.Boolean(True)
468
762
763
469
764
class DenyByDefaultCmd(PropertyCmd):
470
property = "ApprovedByDefault"
765
propname = "ApprovedByDefault"
471
766
value_to_set = dbus.Boolean(False)
472
767
473
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
474
property = "Checker"
475
476
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
477
property = "Host"
478
479
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
768
769
class PropertyValueCmd(PropertyCmd):
770
"""Abstract class for PropertyCmd recieving a value as argument"""
771
def __init__(self, value):
772
self.value_to_set = value
773
774
775
class SetCheckerCmd(PropertyValueCmd):
776
propname = "Checker"
777
778
779
class SetHostCmd(PropertyValueCmd):
780
propname = "Host"
781
782
783
class SetSecretCmd(PropertyValueCmd):
784
propname = "Secret"
785
480
786
@property
481
787
def value_to_set(self):
482
788
return self._vts
789
483
790
@value_to_set.setter
484
791
def value_to_set(self, value):
485
792
"""When setting, read data from supplied file object"""
486
793
self._vts = value.read()
487
794
value.close()
488
property = "Secret"
489
490
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
491
property = "Timeout"
492
493
class SetExtendedTimeoutCmd(PropertyCmd,
494
MillisecondsValueArgumentMixIn):
495
property = "ExtendedTimeout"
496
497
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
498
property = "Interval"
499
500
class SetApprovalDelayCmd(PropertyCmd,
501
MillisecondsValueArgumentMixIn):
502
property = "ApprovalDelay"
503
504
class SetApprovalDurationCmd(PropertyCmd,
505
MillisecondsValueArgumentMixIn):
506
property = "ApprovalDuration"
507
508
def add_command_line_options(parser):
509
parser.add_argument("--version", action="version",
510
version="%(prog)s {}".format(version),
511
help="show version number and exit")
512
parser.add_argument("-a", "--all", action="store_true",
513
help="Select all clients")
514
parser.add_argument("-v", "--verbose", action="store_true",
515
help="Print all fields")
516
parser.add_argument("-j", "--dump-json", action="store_true",
517
help="Dump client data in JSON format")
518
enable_disable = parser.add_mutually_exclusive_group()
519
enable_disable.add_argument("-e", "--enable", action="store_true",
520
help="Enable client")
521
enable_disable.add_argument("-d", "--disable",
522
action="store_true",
523
help="disable client")
524
parser.add_argument("-b", "--bump-timeout", action="store_true",
525
help="Bump timeout for client")
526
start_stop_checker = parser.add_mutually_exclusive_group()
527
start_stop_checker.add_argument("--start-checker",
528
action="store_true",
529
help="Start checker for client")
530
start_stop_checker.add_argument("--stop-checker",
531
action="store_true",
532
help="Stop checker for client")
533
parser.add_argument("-V", "--is-enabled", action="store_true",
534
help="Check if client is enabled")
535
parser.add_argument("-r", "--remove", action="store_true",
536
help="Remove client")
537
parser.add_argument("-c", "--checker",
538
help="Set checker command for client")
539
parser.add_argument("-t", "--timeout", type=string_to_delta,
540
help="Set timeout for client")
541
parser.add_argument("--extended-timeout", type=string_to_delta,
542
help="Set extended timeout for client")
543
parser.add_argument("-i", "--interval", type=string_to_delta,
544
help="Set checker interval for client")
545
approve_deny_default = parser.add_mutually_exclusive_group()
546
approve_deny_default.add_argument(
547
"--approve-by-default", action="store_true",
548
default=None, dest="approved_by_default",
549
help="Set client to be approved by default")
550
approve_deny_default.add_argument(
551
"--deny-by-default", action="store_false",
552
dest="approved_by_default",
553
help="Set client to be denied by default")
554
parser.add_argument("--approval-delay", type=string_to_delta,
555
help="Set delay before client approve/deny")
556
parser.add_argument("--approval-duration", type=string_to_delta,
557
help="Set duration of one client approval")
558
parser.add_argument("-H", "--host", help="Set host for client")
559
parser.add_argument("-s", "--secret",
560
type=argparse.FileType(mode="rb"),
561
help="Set password blob (file) for client")
562
approve_deny = parser.add_mutually_exclusive_group()
563
approve_deny.add_argument(
564
"-A", "--approve", action="store_true",
565
help="Approve any current client request")
566
approve_deny.add_argument("-D", "--deny", action="store_true",
567
help="Deny any current client request")
568
parser.add_argument("--check", action="store_true",
569
help="Run self-test")
570
parser.add_argument("client", nargs="*", help="Client name")
571
572
573
def commands_from_options(options):
574
575
commands = []
576
577
if options.dump_json:
578
commands.append(DumpJSONCmd())
579
580
if options.enable:
581
commands.append(EnableCmd())
582
583
if options.disable:
584
commands.append(DisableCmd())
585
586
if options.bump_timeout:
587
commands.append(BumpTimeoutCmd())
588
589
if options.start_checker:
590
commands.append(StartCheckerCmd())
591
592
if options.stop_checker:
593
commands.append(StopCheckerCmd())
594
595
if options.is_enabled:
596
commands.append(IsEnabledCmd())
597
598
if options.remove:
599
commands.append(RemoveCmd())
600
601
if options.checker is not None:
602
commands.append(SetCheckerCmd(options.checker))
603
604
if options.timeout is not None:
605
commands.append(SetTimeoutCmd(options.timeout))
606
607
if options.extended_timeout:
608
commands.append(
609
SetExtendedTimeoutCmd(options.extended_timeout))
610
611
if options.interval is not None:
612
commands.append(SetIntervalCmd(options.interval))
613
614
if options.approved_by_default is not None:
615
if options.approved_by_default:
616
commands.append(ApproveByDefaultCmd())
617
else:
618
commands.append(DenyByDefaultCmd())
619
620
if options.approval_delay is not None:
621
commands.append(SetApprovalDelayCmd(options.approval_delay))
622
623
if options.approval_duration is not None:
624
commands.append(
625
SetApprovalDurationCmd(options.approval_duration))
626
627
if options.host is not None:
628
commands.append(SetHostCmd(options.host))
629
630
if options.secret is not None:
631
commands.append(SetSecretCmd(options.secret))
632
633
if options.approve:
634
commands.append(ApproveCmd())
635
636
if options.deny:
637
commands.append(DenyCmd())
638
639
# If no command option has been given, show table of clients,
640
# optionally verbosely
641
if not commands:
642
commands.append(PrintTableCmd(verbose=options.verbose))
643
644
return commands
645
646
647
def check_option_syntax(parser, options):
648
649
def has_actions(options):
650
return any((options.enable,
651
options.disable,
652
options.bump_timeout,
653
options.start_checker,
654
options.stop_checker,
655
options.is_enabled,
656
options.remove,
657
options.checker is not None,
658
options.timeout is not None,
659
options.extended_timeout is not None,
660
options.interval is not None,
661
options.approved_by_default is not None,
662
options.approval_delay is not None,
663
options.approval_duration is not None,
664
options.host is not None,
665
options.secret is not None,
666
options.approve,
667
options.deny))
668
669
if has_actions(options) and not (options.client or options.all):
670
parser.error("Options require clients names or --all.")
671
if options.verbose and has_actions(options):
672
parser.error("--verbose can only be used alone.")
673
if options.dump_json and (options.verbose
674
or has_actions(options)):
675
parser.error("--dump-json can only be used alone.")
676
if options.all and not has_actions(options):
677
parser.error("--all requires an action.")
678
if options.is_enabled and len(options.client) > 1:
679
parser.error("--is-enabled requires exactly one client")
680
681
682
def main():
683
parser = argparse.ArgumentParser()
684
685
add_command_line_options(parser)
686
687
options = parser.parse_args()
688
689
check_option_syntax(parser, options)
690
691
clientnames = options.client
692
693
try:
694
bus = dbus.SystemBus()
695
mandos_dbus_objc = bus.get_object(busname, server_path)
696
except dbus.exceptions.DBusException:
697
log.critical("Could not connect to Mandos server")
698
sys.exit(1)
699
700
mandos_serv = dbus.Interface(mandos_dbus_objc,
701
dbus_interface=server_interface)
702
mandos_serv_object_manager = dbus.Interface(
703
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
704
705
# Filter out log message from dbus module
706
dbus_logger = logging.getLogger("dbus.proxies")
707
class NullFilter(logging.Filter):
708
def filter(self, record):
709
return False
710
dbus_filter = NullFilter()
711
try:
712
dbus_logger.addFilter(dbus_filter)
713
mandos_clients = {path: ifs_and_props[client_interface]
714
for path, ifs_and_props in
715
mandos_serv_object_manager
716
.GetManagedObjects().items()
717
if client_interface in ifs_and_props}
718
except dbus.exceptions.DBusException as e:
719
log.critical("Failed to access Mandos server through D-Bus:"
720
"\n%s", e)
721
sys.exit(1)
722
finally:
723
# restore dbus logger
724
dbus_logger.removeFilter(dbus_filter)
725
726
# Compile dict of (clients: properties) to process
727
clients = {}
728
729
if not clientnames:
730
clients = {bus.get_object(busname, path): properties
731
for path, properties in mandos_clients.items()}
732
else:
733
for name in clientnames:
734
for path, client in mandos_clients.items():
735
if client["Name"] == name:
736
client_objc = bus.get_object(busname, path)
737
clients[client_objc] = client
738
break
739
else:
740
log.critical("Client not found on server: %r", name)
741
sys.exit(1)
742
743
# Run all commands on clients
744
commands = commands_from_options(options)
745
for command in commands:
746
command.run(mandos_serv, clients)
795
796
797
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
798
"""Abstract class for PropertyValueCmd taking a value argument as
799
a datetime.timedelta() but should store it as milliseconds."""
800
801
@property
802
def value_to_set(self):
803
return self._vts
804
805
@value_to_set.setter
806
def value_to_set(self, value):
807
"""When setting, convert value from a datetime.timedelta"""
808
self._vts = int(round(value.total_seconds() * 1000))
809
810
811
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
812
propname = "Timeout"
813
814
815
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
816
propname = "ExtendedTimeout"
817
818
819
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
820
propname = "Interval"
821
822
823
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
824
propname = "ApprovalDelay"
825
826
827
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
828
propname = "ApprovalDuration"
829
747
830
748
831
749
class Test_milliseconds_to_string(unittest.TestCase):
750
def test_all(self):
751
self.assertEqual(milliseconds_to_string(93785000),
752
"1T02:03:05")
753
def test_no_days(self):
754
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
755
def test_all_zero(self):
756
self.assertEqual(milliseconds_to_string(0), "00:00:00")
757
def test_no_fractional_seconds(self):
758
self.assertEqual(milliseconds_to_string(400), "00:00:00")
759
self.assertEqual(milliseconds_to_string(900), "00:00:00")
760
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
761
762
832
class Test_string_to_delta(unittest.TestCase):
763
833
def test_handles_basic_rfc3339(self):
764
834
self.assertEqual(string_to_delta("PT0S"),
769
839
datetime.timedelta(0, 1))
770
840
self.assertEqual(string_to_delta("PT2H"),
771
841
datetime.timedelta(0, 7200))
842
772
843
def test_falls_back_to_pre_1_6_1_with_warning(self):
773
844
# assertLogs only exists in Python 3.4
774
845
if hasattr(self, "assertLogs"):
792
863
self.assertEqual(value, datetime.timedelta(0, 7200))
793
864
794
865
866
class Test_check_option_syntax(unittest.TestCase):
867
def setUp(self):
868
self.parser = argparse.ArgumentParser()
869
add_command_line_options(self.parser)
870
871
def test_actions_requires_client_or_all(self):
872
for action, value in self.actions.items():
873
options = self.parser.parse_args()
874
setattr(options, action, value)
875
with self.assertParseError():
876
self.check_option_syntax(options)
877
878
# This mostly corresponds to the definition from has_actions() in
879
# check_option_syntax()
880
actions = {
881
# The actual values set here are not that important, but we do
882
# at least stick to the correct types, even though they are
883
# never used
884
"enable": True,
885
"disable": True,
886
"bump_timeout": True,
887
"start_checker": True,
888
"stop_checker": True,
889
"is_enabled": True,
890
"remove": True,
891
"checker": "x",
892
"timeout": datetime.timedelta(),
893
"extended_timeout": datetime.timedelta(),
894
"interval": datetime.timedelta(),
895
"approved_by_default": True,
896
"approval_delay": datetime.timedelta(),
897
"approval_duration": datetime.timedelta(),
898
"host": "x",
899
"secret": io.BytesIO(b"x"),
900
"approve": True,
901
"deny": True,
902
}
903
904
@contextlib.contextmanager
905
def assertParseError(self):
906
with self.assertRaises(SystemExit) as e:
907
with self.temporarily_suppress_stderr():
908
yield
909
# Exit code from argparse is guaranteed to be "2". Reference:
910
# https://docs.python.org/3/library
911
# /argparse.html#exiting-methods
912
self.assertEqual(e.exception.code, 2)
913
914
@staticmethod
915
@contextlib.contextmanager
916
def temporarily_suppress_stderr():
917
null = os.open(os.path.devnull, os.O_RDWR)
918
stderrcopy = os.dup(sys.stderr.fileno())
919
os.dup2(null, sys.stderr.fileno())
920
os.close(null)
921
try:
922
yield
923
finally:
924
# restore stderr
925
os.dup2(stderrcopy, sys.stderr.fileno())
926
os.close(stderrcopy)
927
928
def check_option_syntax(self, options):
929
check_option_syntax(self.parser, options)
930
931
def test_actions_conflicts_with_verbose(self):
932
for action, value in self.actions.items():
933
options = self.parser.parse_args()
934
setattr(options, action, value)
935
options.verbose = True
936
with self.assertParseError():
937
self.check_option_syntax(options)
938
939
def test_dump_json_conflicts_with_verbose(self):
940
options = self.parser.parse_args()
941
options.dump_json = True
942
options.verbose = True
943
with self.assertParseError():
944
self.check_option_syntax(options)
945
946
def test_dump_json_conflicts_with_action(self):
947
for action, value in self.actions.items():
948
options = self.parser.parse_args()
949
setattr(options, action, value)
950
options.dump_json = True
951
with self.assertParseError():
952
self.check_option_syntax(options)
953
954
def test_all_can_not_be_alone(self):
955
options = self.parser.parse_args()
956
options.all = True
957
with self.assertParseError():
958
self.check_option_syntax(options)
959
960
def test_all_is_ok_with_any_action(self):
961
for action, value in self.actions.items():
962
options = self.parser.parse_args()
963
setattr(options, action, value)
964
options.all = True
965
self.check_option_syntax(options)
966
967
def test_is_enabled_fails_without_client(self):
968
options = self.parser.parse_args()
969
options.is_enabled = True
970
with self.assertParseError():
971
self.check_option_syntax(options)
972
973
def test_is_enabled_works_with_one_client(self):
974
options = self.parser.parse_args()
975
options.is_enabled = True
976
options.client = ["foo"]
977
self.check_option_syntax(options)
978
979
def test_is_enabled_fails_with_two_clients(self):
980
options = self.parser.parse_args()
981
options.is_enabled = True
982
options.client = ["foo", "barbar"]
983
with self.assertParseError():
984
self.check_option_syntax(options)
985
986
def test_remove_can_only_be_combined_with_action_deny(self):
987
for action, value in self.actions.items():
988
if action in {"remove", "deny"}:
989
continue
990
options = self.parser.parse_args()
991
setattr(options, action, value)
992
options.all = True
993
options.remove = True
994
with self.assertParseError():
995
self.check_option_syntax(options)
996
997
998
class Test_get_mandos_dbus_object(unittest.TestCase):
999
def test_calls_and_returns_get_object_on_bus(self):
1000
class MockBus(object):
1001
called = False
1002
def get_object(mockbus_self, busname, dbus_path):
1003
# Note that "self" is still the testcase instance,
1004
# this MockBus instance is in "mockbus_self".
1005
self.assertEqual(busname, dbus_busname)
1006
self.assertEqual(dbus_path, server_dbus_path)
1007
mockbus_self.called = True
1008
return mockbus_self
1009
1010
mockbus = get_mandos_dbus_object(bus=MockBus())
1011
self.assertIsInstance(mockbus, MockBus)
1012
self.assertTrue(mockbus.called)
1013
1014
def test_logs_and_exits_on_dbus_error(self):
1015
class MockBusFailing(object):
1016
def get_object(self, busname, dbus_path):
1017
raise dbus.exceptions.DBusException("Test")
1018
1019
# assertLogs only exists in Python 3.4
1020
if hasattr(self, "assertLogs"):
1021
with self.assertLogs(log, logging.CRITICAL):
1022
with self.assertRaises(SystemExit) as e:
1023
bus = get_mandos_dbus_object(bus=MockBus())
1024
else:
1025
critical_filter = self.CriticalFilter()
1026
log.addFilter(critical_filter)
1027
try:
1028
with self.assertRaises(SystemExit) as e:
1029
get_mandos_dbus_object(bus=MockBusFailing())
1030
finally:
1031
log.removeFilter(critical_filter)
1032
self.assertTrue(critical_filter.found)
1033
if isinstance(e.exception.code, int):
1034
self.assertNotEqual(e.exception.code, 0)
1035
else:
1036
self.assertIsNotNone(e.exception.code)
1037
1038
class CriticalFilter(logging.Filter):
1039
"""Don't show, but register, critical messages"""
1040
found = False
1041
def filter(self, record):
1042
is_critical = record.levelno >= logging.CRITICAL
1043
self.found = is_critical or self.found
1044
return not is_critical
1045
1046
1047
class Test_SilenceLogger(unittest.TestCase):
1048
loggername = "mandos-ctl.Test_SilenceLogger"
1049
log = logging.getLogger(loggername)
1050
log.propagate = False
1051
log.addHandler(logging.NullHandler())
1052
1053
def setUp(self):
1054
self.counting_filter = self.CountingFilter()
1055
1056
class CountingFilter(logging.Filter):
1057
"Count number of records"
1058
count = 0
1059
def filter(self, record):
1060
self.count += 1
1061
return True
1062
1063
def test_should_filter_records_only_when_active(self):
1064
try:
1065
with SilenceLogger(self.loggername):
1066
self.log.addFilter(self.counting_filter)
1067
self.log.info("Filtered log message 1")
1068
self.log.info("Non-filtered message 2")
1069
self.log.info("Non-filtered message 3")
1070
finally:
1071
self.log.removeFilter(self.counting_filter)
1072
self.assertEqual(self.counting_filter.count, 2)
1073
1074
1075
class Test_commands_from_options(unittest.TestCase):
1076
def setUp(self):
1077
self.parser = argparse.ArgumentParser()
1078
add_command_line_options(self.parser)
1079
1080
def test_is_enabled(self):
1081
self.assert_command_from_args(["--is-enabled", "foo"],
1082
IsEnabledCmd)
1083
1084
def assert_command_from_args(self, args, command_cls,
1085
**cmd_attrs):
1086
"""Assert that parsing ARGS should result in an instance of
1087
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1088
options = self.parser.parse_args(args)
1089
check_option_syntax(self.parser, options)
1090
commands = commands_from_options(options)
1091
self.assertEqual(len(commands), 1)
1092
command = commands[0]
1093
self.assertIsInstance(command, command_cls)
1094
for key, value in cmd_attrs.items():
1095
self.assertEqual(getattr(command, key), value)
1096
1097
def test_is_enabled_short(self):
1098
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1099
1100
def test_approve(self):
1101
self.assert_command_from_args(["--approve", "foo"],
1102
ApproveCmd)
1103
1104
def test_approve_short(self):
1105
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1106
1107
def test_deny(self):
1108
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1109
1110
def test_deny_short(self):
1111
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1112
1113
def test_remove(self):
1114
self.assert_command_from_args(["--remove", "foo"],
1115
RemoveCmd)
1116
1117
def test_deny_before_remove(self):
1118
options = self.parser.parse_args(["--deny", "--remove",
1119
"foo"])
1120
check_option_syntax(self.parser, options)
1121
commands = commands_from_options(options)
1122
self.assertEqual(len(commands), 2)
1123
self.assertIsInstance(commands[0], DenyCmd)
1124
self.assertIsInstance(commands[1], RemoveCmd)
1125
1126
def test_deny_before_remove_reversed(self):
1127
options = self.parser.parse_args(["--remove", "--deny",
1128
"--all"])
1129
check_option_syntax(self.parser, options)
1130
commands = commands_from_options(options)
1131
self.assertEqual(len(commands), 2)
1132
self.assertIsInstance(commands[0], DenyCmd)
1133
self.assertIsInstance(commands[1], RemoveCmd)
1134
1135
def test_remove_short(self):
1136
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1137
1138
def test_dump_json(self):
1139
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1140
1141
def test_enable(self):
1142
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1143
1144
def test_enable_short(self):
1145
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1146
1147
def test_disable(self):
1148
self.assert_command_from_args(["--disable", "foo"],
1149
DisableCmd)
1150
1151
def test_disable_short(self):
1152
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1153
1154
def test_bump_timeout(self):
1155
self.assert_command_from_args(["--bump-timeout", "foo"],
1156
BumpTimeoutCmd)
1157
1158
def test_bump_timeout_short(self):
1159
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1160
1161
def test_start_checker(self):
1162
self.assert_command_from_args(["--start-checker", "foo"],
1163
StartCheckerCmd)
1164
1165
def test_stop_checker(self):
1166
self.assert_command_from_args(["--stop-checker", "foo"],
1167
StopCheckerCmd)
1168
1169
def test_approve_by_default(self):
1170
self.assert_command_from_args(["--approve-by-default", "foo"],
1171
ApproveByDefaultCmd)
1172
1173
def test_deny_by_default(self):
1174
self.assert_command_from_args(["--deny-by-default", "foo"],
1175
DenyByDefaultCmd)
1176
1177
def test_checker(self):
1178
self.assert_command_from_args(["--checker", ":", "foo"],
1179
SetCheckerCmd, value_to_set=":")
1180
1181
def test_checker_empty(self):
1182
self.assert_command_from_args(["--checker", "", "foo"],
1183
SetCheckerCmd, value_to_set="")
1184
1185
def test_checker_short(self):
1186
self.assert_command_from_args(["-c", ":", "foo"],
1187
SetCheckerCmd, value_to_set=":")
1188
1189
def test_host(self):
1190
self.assert_command_from_args(["--host", "foo.example.org",
1191
"foo"], SetHostCmd,
1192
value_to_set="foo.example.org")
1193
1194
def test_host_short(self):
1195
self.assert_command_from_args(["-H", "foo.example.org",
1196
"foo"], SetHostCmd,
1197
value_to_set="foo.example.org")
1198
1199
def test_secret_devnull(self):
1200
self.assert_command_from_args(["--secret", os.path.devnull,
1201
"foo"], SetSecretCmd,
1202
value_to_set=b"")
1203
1204
def test_secret_tempfile(self):
1205
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1206
value = b"secret\0xyzzy\nbar"
1207
f.write(value)
1208
f.seek(0)
1209
self.assert_command_from_args(["--secret", f.name,
1210
"foo"], SetSecretCmd,
1211
value_to_set=value)
1212
1213
def test_secret_devnull_short(self):
1214
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1215
SetSecretCmd, value_to_set=b"")
1216
1217
def test_secret_tempfile_short(self):
1218
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1219
value = b"secret\0xyzzy\nbar"
1220
f.write(value)
1221
f.seek(0)
1222
self.assert_command_from_args(["-s", f.name, "foo"],
1223
SetSecretCmd,
1224
value_to_set=value)
1225
1226
def test_timeout(self):
1227
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1228
SetTimeoutCmd,
1229
value_to_set=300000)
1230
1231
def test_timeout_short(self):
1232
self.assert_command_from_args(["-t", "PT5M", "foo"],
1233
SetTimeoutCmd,
1234
value_to_set=300000)
1235
1236
def test_extended_timeout(self):
1237
self.assert_command_from_args(["--extended-timeout", "PT15M",
1238
"foo"],
1239
SetExtendedTimeoutCmd,
1240
value_to_set=900000)
1241
1242
def test_interval(self):
1243
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1244
SetIntervalCmd,
1245
value_to_set=120000)
1246
1247
def test_interval_short(self):
1248
self.assert_command_from_args(["-i", "PT2M", "foo"],
1249
SetIntervalCmd,
1250
value_to_set=120000)
1251
1252
def test_approval_delay(self):
1253
self.assert_command_from_args(["--approval-delay", "PT30S",
1254
"foo"], SetApprovalDelayCmd,
1255
value_to_set=30000)
1256
1257
def test_approval_duration(self):
1258
self.assert_command_from_args(["--approval-duration", "PT1S",
1259
"foo"], SetApprovalDurationCmd,
1260
value_to_set=1000)
1261
1262
def test_print_table(self):
1263
self.assert_command_from_args([], PrintTableCmd,
1264
verbose=False)
1265
1266
def test_print_table_verbose(self):
1267
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1268
verbose=True)
1269
1270
def test_print_table_verbose_short(self):
1271
self.assert_command_from_args(["-v"], PrintTableCmd,
1272
verbose=True)
1273
1274
795
1275
class TestCmd(unittest.TestCase):
796
1276
"""Abstract class for tests of command classes"""
1277
797
1278
def setUp(self):
798
1279
testcase = self
799
1280
class MockClient(object):
800
1281
def __init__(self, name, **attributes):
801
self.__dbus_object_path__ = "objpath_{}".format(name)
1282
self.__dbus_object_path__ = "/clients/{}".format(name)
802
1283
self.attributes = attributes
803
1284
self.attributes["Name"] = name
804
1285
self.calls = []
805
def Set(self, interface, property, value, dbus_interface):
806
testcase.assertEqual(interface, client_interface)
807
testcase.assertEqual(dbus_interface,
808
dbus.PROPERTIES_IFACE)
809
self.attributes[property] = value
810
def Get(self, interface, property, dbus_interface):
811
testcase.assertEqual(interface, client_interface)
812
testcase.assertEqual(dbus_interface,
813
dbus.PROPERTIES_IFACE)
814
return self.attributes[property]
1286
def Set(self, interface, propname, value, dbus_interface):
1287
testcase.assertEqual(interface, client_dbus_interface)
1288
testcase.assertEqual(dbus_interface,
1289
dbus.PROPERTIES_IFACE)
1290
self.attributes[propname] = value
1291
def Get(self, interface, propname, dbus_interface):
1292
testcase.assertEqual(interface, client_dbus_interface)
1293
testcase.assertEqual(dbus_interface,
1294
dbus.PROPERTIES_IFACE)
1295
return self.attributes[propname]
815
1296
def Approve(self, approve, dbus_interface):
816
testcase.assertEqual(dbus_interface, client_interface)
1297
testcase.assertEqual(dbus_interface,
1298
client_dbus_interface)
817
1299
self.calls.append(("Approve", (approve,
818
1300
dbus_interface)))
819
1301
self.client = MockClient(
859
1341
ApprovedByDefault=dbus.Boolean(False),
860
1342
LastApprovalRequest="2019-01-03T00:00:00",
861
1343
ApprovalDelay=30000,
862
ApprovalDuration=1000,
1344
ApprovalDuration=93785000,
863
1345
Checker=":",
864
1346
ExtendedTimeout=900000,
865
1347
Expires="2019-02-05T00:00:00",
866
1348
LastCheckerStatus=-2)
867
1349
self.clients = collections.OrderedDict(
868
1350
[
869
(self.client, self.client.attributes),
870
(self.other_client, self.other_client.attributes),
1351
("/clients/foo", self.client.attributes),
1352
("/clients/barbar", self.other_client.attributes),
871
1353
])
872
self.one_client = {self.client: self.client.attributes}
873
874
class TestPrintTableCmd(TestCmd):
875
def test_normal(self):
876
output = PrintTableCmd().output(self.clients)
877
expected_output = """
878
Name Enabled Timeout Last Successful Check
879
foo Yes 00:05:00 2019-02-03T00:00:00
880
barbar Yes 00:05:00 2019-02-04T00:00:00
881
"""[1:-1]
882
self.assertEqual(output, expected_output)
883
def test_verbose(self):
884
output = PrintTableCmd(verbose=True).output(self.clients)
885
expected_output = """
886
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
887
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
888
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
889
"""[1:-1]
890
self.assertEqual(output, expected_output)
891
def test_one_client(self):
892
output = PrintTableCmd().output(self.one_client)
893
expected_output = """
894
Name Enabled Timeout Last Successful Check
895
foo Yes 00:05:00 2019-02-03T00:00:00
896
"""[1:-1]
897
self.assertEqual(output, expected_output)
1354
self.one_client = {"/clients/foo": self.client.attributes}
1355
1356
@property
1357
def bus(self):
1358
class Bus(object):
1359
@staticmethod
1360
def get_object(client_bus_name, path):
1361
self.assertEqual(client_bus_name, dbus_busname)
1362
return {
1363
# Note: "self" here is the TestCmd instance, not
1364
# the Bus instance, since this is a static method!
1365
"/clients/foo": self.client,
1366
"/clients/barbar": self.other_client,
1367
}[path]
1368
return Bus()
1369
1370
1371
class TestIsEnabledCmd(TestCmd):
1372
def test_is_enabled(self):
1373
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1374
properties)
1375
for client, properties
1376
in self.clients.items()))
1377
1378
def test_is_enabled_run_exits_successfully(self):
1379
with self.assertRaises(SystemExit) as e:
1380
IsEnabledCmd().run(self.one_client)
1381
if e.exception.code is not None:
1382
self.assertEqual(e.exception.code, 0)
1383
else:
1384
self.assertIsNone(e.exception.code)
1385
1386
def test_is_enabled_run_exits_with_failure(self):
1387
self.client.attributes["Enabled"] = dbus.Boolean(False)
1388
with self.assertRaises(SystemExit) as e:
1389
IsEnabledCmd().run(self.one_client)
1390
if isinstance(e.exception.code, int):
1391
self.assertNotEqual(e.exception.code, 0)
1392
else:
1393
self.assertIsNotNone(e.exception.code)
1394
1395
1396
class TestApproveCmd(TestCmd):
1397
def test_approve(self):
1398
ApproveCmd().run(self.clients, self.bus)
1399
for clientpath in self.clients:
1400
client = self.bus.get_object(dbus_busname, clientpath)
1401
self.assertIn(("Approve", (True, client_dbus_interface)),
1402
client.calls)
1403
1404
1405
class TestDenyCmd(TestCmd):
1406
def test_deny(self):
1407
DenyCmd().run(self.clients, self.bus)
1408
for clientpath in self.clients:
1409
client = self.bus.get_object(dbus_busname, clientpath)
1410
self.assertIn(("Approve", (False, client_dbus_interface)),
1411
client.calls)
1412
1413
1414
class TestRemoveCmd(TestCmd):
1415
def test_remove(self):
1416
class MockMandos(object):
1417
def __init__(self):
1418
self.calls = []
1419
def RemoveClient(self, dbus_path):
1420
self.calls.append(("RemoveClient", (dbus_path,)))
1421
mandos = MockMandos()
1422
super(TestRemoveCmd, self).setUp()
1423
RemoveCmd().run(self.clients, self.bus, mandos)
1424
self.assertEqual(len(mandos.calls), 2)
1425
for clientpath in self.clients:
1426
self.assertIn(("RemoveClient", (clientpath,)),
1427
mandos.calls)
1428
898
1429
899
1430
class TestDumpJSONCmd(TestCmd):
900
1431
def setUp(self):
941
1472
"ApprovedByDefault": False,
942
1473
"LastApprovalRequest": "2019-01-03T00:00:00",
943
1474
"ApprovalDelay": 30000,
944
"ApprovalDuration": 1000,
1475
"ApprovalDuration": 93785000,
945
1476
"Checker": ":",
946
1477
"ExtendedTimeout": 900000,
947
1478
"Expires": "2019-02-05T00:00:00",
949
1480
},
950
1481
}
951
1482
return super(TestDumpJSONCmd, self).setUp()
1483
952
1484
def test_normal(self):
953
json_data = json.loads(DumpJSONCmd().output(self.clients))
1485
output = DumpJSONCmd().output(self.clients.values())
1486
json_data = json.loads(output)
954
1487
self.assertDictEqual(json_data, self.expected_json)
1488
955
1489
def test_one_client(self):
956
clients = self.one_client
957
json_data = json.loads(DumpJSONCmd().output(clients))
1490
output = DumpJSONCmd().output(self.one_client.values())
1491
json_data = json.loads(output)
958
1492
expected_json = {"foo": self.expected_json["foo"]}
959
1493
self.assertDictEqual(json_data, expected_json)
960
1494
961
class TestIsEnabledCmd(TestCmd):
962
def test_is_enabled(self):
963
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
964
for client, properties in self.clients.items()))
965
def test_is_enabled_run_exits_successfully(self):
966
with self.assertRaises(SystemExit) as e:
967
IsEnabledCmd().run(None, self.one_client)
968
if e.exception.code is not None:
969
self.assertEqual(e.exception.code, 0)
970
else:
971
self.assertIsNone(e.exception.code)
972
def test_is_enabled_run_exits_with_failure(self):
973
self.client.attributes["Enabled"] = dbus.Boolean(False)
974
with self.assertRaises(SystemExit) as e:
975
IsEnabledCmd().run(None, self.one_client)
976
if isinstance(e.exception.code, int):
977
self.assertNotEqual(e.exception.code, 0)
978
else:
979
self.assertIsNotNone(e.exception.code)
980
981
class TestRemoveCmd(TestCmd):
982
def test_remove(self):
983
class MockMandos(object):
984
def __init__(self):
985
self.calls = []
986
def RemoveClient(self, dbus_path):
987
self.calls.append(("RemoveClient", (dbus_path,)))
988
mandos = MockMandos()
989
super(TestRemoveCmd, self).setUp()
990
RemoveCmd().run(mandos, self.clients)
991
self.assertEqual(len(mandos.calls), 2)
992
for client in self.clients:
993
self.assertIn(("RemoveClient",
994
(client.__dbus_object_path__,)),
995
mandos.calls)
996
997
class TestApproveCmd(TestCmd):
998
def test_approve(self):
999
ApproveCmd().run(None, self.clients)
1000
for client in self.clients:
1001
self.assertIn(("Approve", (True, client_interface)),
1002
client.calls)
1003
1004
class TestDenyCmd(TestCmd):
1005
def test_deny(self):
1006
DenyCmd().run(None, self.clients)
1007
for client in self.clients:
1008
self.assertIn(("Approve", (False, client_interface)),
1009
client.calls)
1010
1011
class TestEnableCmd(TestCmd):
1012
def test_enable(self):
1013
for client in self.clients:
1014
client.attributes["Enabled"] = False
1015
1016
EnableCmd().run(None, self.clients)
1017
1018
for client in self.clients:
1019
self.assertTrue(client.attributes["Enabled"])
1020
1021
class TestDisableCmd(TestCmd):
1022
def test_disable(self):
1023
DisableCmd().run(None, self.clients)
1024
1025
for client in self.clients:
1026
self.assertFalse(client.attributes["Enabled"])
1027
1028
class Unique(object):
1029
"""Class for objects which exist only to be unique objects, since
1030
unittest.mock.sentinel only exists in Python 3.3"""
1495
1496
class TestPrintTableCmd(TestCmd):
1497
def test_normal(self):
1498
output = PrintTableCmd().output(self.clients.values())
1499
expected_output = "\n".join((
1500
"Name Enabled Timeout Last Successful Check",
1501
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1502
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1503
))
1504
self.assertEqual(output, expected_output)
1505
1506
def test_verbose(self):
1507
output = PrintTableCmd(verbose=True).output(
1508
self.clients.values())
1509
columns = (
1510
(
1511
"Name ",
1512
"foo ",
1513
"barbar ",
1514
),(
1515
"Enabled ",
1516
"Yes ",
1517
"Yes ",
1518
),(
1519
"Timeout ",
1520
"00:05:00 ",
1521
"00:05:00 ",
1522
),(
1523
"Last Successful Check ",
1524
"2019-02-03T00:00:00 ",
1525
"2019-02-04T00:00:00 ",
1526
),(
1527
"Created ",
1528
"2019-01-02T00:00:00 ",
1529
"2019-01-03T00:00:00 ",
1530
),(
1531
"Interval ",
1532
"00:02:00 ",
1533
"00:02:00 ",
1534
),(
1535
"Host ",
1536
"foo.example.org ",
1537
"192.0.2.3 ",
1538
),(
1539
("Key ID "
1540
" "),
1541
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1542
"ea90a27cddb12 "),
1543
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1544
"46c2b020f441c "),
1545
),(
1546
"Fingerprint ",
1547
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1548
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1549
),(
1550
"Check Is Running ",
1551
"No ",
1552
"Yes ",
1553
),(
1554
"Last Enabled ",
1555
"2019-01-03T00:00:00 ",
1556
"2019-01-04T00:00:00 ",
1557
),(
1558
"Approval Is Pending ",
1559
"No ",
1560
"No ",
1561
),(
1562
"Approved By Default ",
1563
"Yes ",
1564
"No ",
1565
),(
1566
"Last Approval Request ",
1567
" ",
1568
"2019-01-03T00:00:00 ",
1569
),(
1570
"Approval Delay ",
1571
"00:00:00 ",
1572
"00:00:30 ",
1573
),(
1574
"Approval Duration ",
1575
"00:00:01 ",
1576
"1T02:03:05 ",
1577
),(
1578
"Checker ",
1579
"fping -q -- %(host)s ",
1580
": ",
1581
),(
1582
"Extended Timeout ",
1583
"00:15:00 ",
1584
"00:15:00 ",
1585
),(
1586
"Expires ",
1587
"2019-02-04T00:00:00 ",
1588
"2019-02-05T00:00:00 ",
1589
),(
1590
"Last Checker Status",
1591
"0 ",
1592
"-2 ",
1593
)
1594
)
1595
num_lines = max(len(rows) for rows in columns)
1596
expected_output = "\n".join("".join(rows[line]
1597
for rows in columns)
1598
for line in range(num_lines))
1599
self.assertEqual(output, expected_output)
1600
1601
def test_one_client(self):
1602
output = PrintTableCmd().output(self.one_client.values())
1603
expected_output = "\n".join((
1604
"Name Enabled Timeout Last Successful Check",
1605
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1606
))
1607
self.assertEqual(output, expected_output)
1608
1031
1609
1032
1610
class TestPropertyCmd(TestCmd):
1033
1611
"""Abstract class for tests of PropertyCmd classes"""
1038
1616
self.values_to_set)
1039
1617
for value_to_set, value_to_get in zip(self.values_to_set,
1040
1618
values_to_get):
1041
for client in self.clients:
1042
old_value = client.attributes[self.property]
1043
self.assertNotIsInstance(old_value, Unique)
1044
client.attributes[self.property] = Unique()
1619
for clientpath in self.clients:
1620
client = self.bus.get_object(dbus_busname, clientpath)
1621
old_value = client.attributes[self.propname]
1622
self.assertNotIsInstance(old_value, self.Unique)
1623
client.attributes[self.propname] = self.Unique()
1045
1624
self.run_command(value_to_set, self.clients)
1046
for client in self.clients:
1047
value = client.attributes[self.property]
1048
self.assertNotIsInstance(value, Unique)
1625
for clientpath in self.clients:
1626
client = self.bus.get_object(dbus_busname, clientpath)
1627
value = client.attributes[self.propname]
1628
self.assertNotIsInstance(value, self.Unique)
1049
1629
self.assertEqual(value, value_to_get)
1630
1631
class Unique(object):
1632
"""Class for objects which exist only to be unique objects,
1633
since unittest.mock.sentinel only exists in Python 3.3"""
1634
1050
1635
def run_command(self, value, clients):
1051
self.command().run(None, clients)
1636
self.command().run(clients, self.bus)
1637
1638
1639
class TestEnableCmd(TestPropertyCmd):
1640
command = EnableCmd
1641
propname = "Enabled"
1642
values_to_set = [dbus.Boolean(True)]
1643
1644
1645
class TestDisableCmd(TestPropertyCmd):
1646
command = DisableCmd
1647
propname = "Enabled"
1648
values_to_set = [dbus.Boolean(False)]
1649
1052
1650
1053
1651
class TestBumpTimeoutCmd(TestPropertyCmd):
1054
1652
command = BumpTimeoutCmd
1055
property = "LastCheckedOK"
1653
propname = "LastCheckedOK"
1056
1654
values_to_set = [""]
1057
1655
1656
1058
1657
class TestStartCheckerCmd(TestPropertyCmd):
1059
1658
command = StartCheckerCmd
1060
property = "CheckerRunning"
1659
propname = "CheckerRunning"
1061
1660
values_to_set = [dbus.Boolean(True)]
1062
1661
1662
1063
1663
class TestStopCheckerCmd(TestPropertyCmd):
1064
1664
command = StopCheckerCmd
1065
property = "CheckerRunning"
1665
propname = "CheckerRunning"
1066
1666
values_to_set = [dbus.Boolean(False)]
1067
1667
1668
1068
1669
class TestApproveByDefaultCmd(TestPropertyCmd):
1069
1670
command = ApproveByDefaultCmd
1070
property = "ApprovedByDefault"
1671
propname = "ApprovedByDefault"
1071
1672
values_to_set = [dbus.Boolean(True)]
1072
1673
1674
1073
1675
class TestDenyByDefaultCmd(TestPropertyCmd):
1074
1676
command = DenyByDefaultCmd
1075
property = "ApprovedByDefault"
1677
propname = "ApprovedByDefault"
1076
1678
values_to_set = [dbus.Boolean(False)]
1077
1679
1078
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1079
"""Abstract class for tests of PropertyCmd classes using the
1080
ValueArgumentMixIn"""
1680
1681
class TestPropertyValueCmd(TestPropertyCmd):
1682
"""Abstract class for tests of PropertyValueCmd classes"""
1683
1081
1684
def runTest(self):
1082
if type(self) is TestValueArgumentPropertyCmd:
1685
if type(self) is TestPropertyValueCmd:
1083
1686
return
1084
return super(TestValueArgumentPropertyCmd, self).runTest()
1687
return super(TestPropertyValueCmd, self).runTest()
1688
1085
1689
def run_command(self, value, clients):
1086
self.command(value).run(None, clients)
1087
1088
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1690
self.command(value).run(clients, self.bus)
1691
1692
1693
class TestSetCheckerCmd(TestPropertyValueCmd):
1089
1694
command = SetCheckerCmd
1090
property = "Checker"
1695
propname = "Checker"
1091
1696
values_to_set = ["", ":", "fping -q -- %s"]
1092
1697
1093
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1698
1699
class TestSetHostCmd(TestPropertyValueCmd):
1094
1700
command = SetHostCmd
1095
property = "Host"
1701
propname = "Host"
1096
1702
values_to_set = ["192.0.2.3", "foo.example.org"]
1097
1703
1098
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1704
1705
class TestSetSecretCmd(TestPropertyValueCmd):
1099
1706
command = SetSecretCmd
1100
property = "Secret"
1101
values_to_set = [open("/dev/null", "rb"),
1707
propname = "Secret"
1708
values_to_set = [io.BytesIO(b""),
1102
1709
io.BytesIO(b"secret\0xyzzy\nbar")]
1103
1710
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1104
1711
1105
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1712
1713
class TestSetTimeoutCmd(TestPropertyValueCmd):
1106
1714
command = SetTimeoutCmd
1107
property = "Timeout"
1715
propname = "Timeout"
1108
1716
values_to_set = [datetime.timedelta(),
1109
1717
datetime.timedelta(minutes=5),
1110
1718
datetime.timedelta(seconds=1),
1112
1720
datetime.timedelta(weeks=52)]
1113
1721
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1114
1722
1115
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1723
1724
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1116
1725
command = SetExtendedTimeoutCmd
1117
property = "ExtendedTimeout"
1726
propname = "ExtendedTimeout"
1118
1727
values_to_set = [datetime.timedelta(),
1119
1728
datetime.timedelta(minutes=5),
1120
1729
datetime.timedelta(seconds=1),
1122
1731
datetime.timedelta(weeks=52)]
1123
1732
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1124
1733
1125
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1734
1735
class TestSetIntervalCmd(TestPropertyValueCmd):
1126
1736
command = SetIntervalCmd
1127
property = "Interval"
1737
propname = "Interval"
1128
1738
values_to_set = [datetime.timedelta(),
1129
1739
datetime.timedelta(minutes=5),
1130
1740
datetime.timedelta(seconds=1),
1132
1742
datetime.timedelta(weeks=52)]
1133
1743
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1134
1744
1135
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1745
1746
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1136
1747
command = SetApprovalDelayCmd
1137
property = "ApprovalDelay"
1748
propname = "ApprovalDelay"
1138
1749
values_to_set = [datetime.timedelta(),
1139
1750
datetime.timedelta(minutes=5),
1140
1751
datetime.timedelta(seconds=1),
1142
1753
datetime.timedelta(weeks=52)]
1143
1754
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1144
1755
1145
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1756
1757
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1146
1758
command = SetApprovalDurationCmd
1147
property = "ApprovalDuration"
1759
propname = "ApprovalDuration"
1148
1760
values_to_set = [datetime.timedelta(),
1149
1761
datetime.timedelta(minutes=5),
1150
1762
datetime.timedelta(seconds=1),
1152
1764
datetime.timedelta(weeks=52)]
1153
1765
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1154
1766
1155
class Test_command_from_options(unittest.TestCase):
1156
def setUp(self):
1157
self.parser = argparse.ArgumentParser()
1158
add_command_line_options(self.parser)
1159
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1160
"""Assert that parsing ARGS should result in an instance of
1161
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1162
options = self.parser.parse_args(args)
1163
check_option_syntax(self.parser, options)
1164
commands = commands_from_options(options)
1165
self.assertEqual(len(commands), 1)
1166
command = commands[0]
1167
self.assertIsInstance(command, command_cls)
1168
for key, value in cmd_attrs.items():
1169
self.assertEqual(getattr(command, key), value)
1170
def test_print_table(self):
1171
self.assert_command_from_args([], PrintTableCmd,
1172
verbose=False)
1173
1174
def test_print_table_verbose(self):
1175
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1176
verbose=True)
1177
1178
def test_print_table_verbose_short(self):
1179
self.assert_command_from_args(["-v"], PrintTableCmd,
1180
verbose=True)
1181
1182
def test_enable(self):
1183
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1184
1185
def test_enable_short(self):
1186
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1187
1188
def test_disable(self):
1189
self.assert_command_from_args(["--disable", "foo"],
1190
DisableCmd)
1191
1192
def test_disable_short(self):
1193
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1194
1195
def test_bump_timeout(self):
1196
self.assert_command_from_args(["--bump-timeout", "foo"],
1197
BumpTimeoutCmd)
1198
1199
def test_bump_timeout_short(self):
1200
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1201
1202
def test_start_checker(self):
1203
self.assert_command_from_args(["--start-checker", "foo"],
1204
StartCheckerCmd)
1205
1206
def test_stop_checker(self):
1207
self.assert_command_from_args(["--stop-checker", "foo"],
1208
StopCheckerCmd)
1209
1210
def test_remove(self):
1211
self.assert_command_from_args(["--remove", "foo"],
1212
RemoveCmd)
1213
1214
def test_remove_short(self):
1215
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1216
1217
def test_checker(self):
1218
self.assert_command_from_args(["--checker", ":", "foo"],
1219
SetCheckerCmd, value_to_set=":")
1220
1221
def test_checker_empty(self):
1222
self.assert_command_from_args(["--checker", "", "foo"],
1223
SetCheckerCmd, value_to_set="")
1224
1225
def test_checker_short(self):
1226
self.assert_command_from_args(["-c", ":", "foo"],
1227
SetCheckerCmd, value_to_set=":")
1228
1229
def test_timeout(self):
1230
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1231
SetTimeoutCmd,
1232
value_to_set=300000)
1233
1234
def test_timeout_short(self):
1235
self.assert_command_from_args(["-t", "PT5M", "foo"],
1236
SetTimeoutCmd,
1237
value_to_set=300000)
1238
1239
def test_extended_timeout(self):
1240
self.assert_command_from_args(["--extended-timeout", "PT15M",
1241
"foo"],
1242
SetExtendedTimeoutCmd,
1243
value_to_set=900000)
1244
1245
def test_interval(self):
1246
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1247
SetIntervalCmd,
1248
value_to_set=120000)
1249
1250
def test_interval_short(self):
1251
self.assert_command_from_args(["-i", "PT2M", "foo"],
1252
SetIntervalCmd,
1253
value_to_set=120000)
1254
1255
def test_approve_by_default(self):
1256
self.assert_command_from_args(["--approve-by-default", "foo"],
1257
ApproveByDefaultCmd)
1258
1259
def test_deny_by_default(self):
1260
self.assert_command_from_args(["--deny-by-default", "foo"],
1261
DenyByDefaultCmd)
1262
1263
def test_approval_delay(self):
1264
self.assert_command_from_args(["--approval-delay", "PT30S",
1265
"foo"], SetApprovalDelayCmd,
1266
value_to_set=30000)
1267
1268
def test_approval_duration(self):
1269
self.assert_command_from_args(["--approval-duration", "PT1S",
1270
"foo"], SetApprovalDurationCmd,
1271
value_to_set=1000)
1272
1273
def test_host(self):
1274
self.assert_command_from_args(["--host", "foo.example.org",
1275
"foo"], SetHostCmd,
1276
value_to_set="foo.example.org")
1277
1278
def test_host_short(self):
1279
self.assert_command_from_args(["-H", "foo.example.org",
1280
"foo"], SetHostCmd,
1281
value_to_set="foo.example.org")
1282
1283
def test_secret_devnull(self):
1284
self.assert_command_from_args(["--secret", os.path.devnull,
1285
"foo"], SetSecretCmd,
1286
value_to_set=b"")
1287
1288
def test_secret_tempfile(self):
1289
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1290
value = b"secret\0xyzzy\nbar"
1291
f.write(value)
1292
f.seek(0)
1293
self.assert_command_from_args(["--secret", f.name,
1294
"foo"], SetSecretCmd,
1295
value_to_set=value)
1296
1297
def test_secret_devnull_short(self):
1298
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1299
SetSecretCmd, value_to_set=b"")
1300
1301
def test_secret_tempfile_short(self):
1302
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1303
value = b"secret\0xyzzy\nbar"
1304
f.write(value)
1305
f.seek(0)
1306
self.assert_command_from_args(["-s", f.name, "foo"],
1307
SetSecretCmd,
1308
value_to_set=value)
1309
1310
def test_approve(self):
1311
self.assert_command_from_args(["--approve", "foo"],
1312
ApproveCmd)
1313
1314
def test_approve_short(self):
1315
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1316
1317
def test_deny(self):
1318
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1319
1320
def test_deny_short(self):
1321
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1322
1323
def test_dump_json(self):
1324
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1325
1326
def test_is_enabled(self):
1327
self.assert_command_from_args(["--is-enabled", "foo"],
1328
IsEnabledCmd)
1329
1330
def test_is_enabled_short(self):
1331
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1332
1333
1767
1334
1768
1335
1769
def should_only_run_tests():
1349
1783
return tests
1350
1784
1351
1785
if __name__ == "__main__":
1352
if should_only_run_tests():
1353
# Call using ./tdd-python-script --check [--verbose]
1354
unittest.main()
1355
else:
1356
main()
1786
try:
1787
if should_only_run_tests():
1788
# Call using ./tdd-python-script --check [--verbose]
1789
unittest.main()
1790
else:
1791
main()
1792
finally:
1793
logging.shutdown()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0