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