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