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