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 1055
- compare with another revision
- download diff
- download tarball
- view history from revision 1055
- Committer: Teddy Hogeborn
- Date: 2019-03-12 20:13:34 UTC
- Revision ID: teddy@recompile.se-20190312201334-my3htrprewjosuw5
mandos-ctl: Refactor
* mandos-ctl: Reorder everything into logical order; put main() first,
and put every subsequent definition as soon as possible
after its first use, except superclasses which need to
be placed before the classes inheriting from them.
Reorder all tests to match.
* mandos-ctl: Reorder everything into logical order; put main() first,
and put every subsequent definition as soon as possible
after its first use, except superclasses which need to
be placed before the classes inheriting from them.
Reorder all tests to match.
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
#
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.8.2"
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
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)
95
232
96
233
97
234
def rfc3339_duration_to_delta(duration):
103
240
datetime.timedelta(0, 60)
104
241
>>> rfc3339_duration_to_delta("PT60M")
105
242
datetime.timedelta(0, 3600)
243
>>> rfc3339_duration_to_delta("P60M")
244
datetime.timedelta(1680)
106
245
>>> rfc3339_duration_to_delta("PT24H")
107
246
datetime.timedelta(1)
108
247
>>> rfc3339_duration_to_delta("P1W")
111
250
datetime.timedelta(0, 330)
112
251
>>> rfc3339_duration_to_delta("P1DT3M20S")
113
252
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'
114
282
"""
115
283
116
284
# Parsing an RFC 3339 duration with regular expressions is not
193
361
return value
194
362
195
363
196
def string_to_delta(interval):
197
"""Parse a string and return a datetime.timedelta
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
198
367
199
>>> string_to_delta('7d')
368
>>> parse_pre_1_6_1_interval('7d')
200
369
datetime.timedelta(7)
201
>>> string_to_delta('60s')
370
>>> parse_pre_1_6_1_interval('60s')
202
371
datetime.timedelta(0, 60)
203
>>> string_to_delta('60m')
372
>>> parse_pre_1_6_1_interval('60m')
204
373
datetime.timedelta(0, 3600)
205
>>> string_to_delta('24h')
374
>>> parse_pre_1_6_1_interval('24h')
206
375
datetime.timedelta(1)
207
>>> string_to_delta('1w')
376
>>> parse_pre_1_6_1_interval('1w')
208
377
datetime.timedelta(7)
209
>>> string_to_delta('5m 30s')
378
>>> parse_pre_1_6_1_interval('5m 30s')
210
379
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
211
386
"""
212
387
213
try:
214
return rfc3339_duration_to_delta(interval)
215
except ValueError:
216
pass
217
218
388
value = datetime.timedelta(0)
219
389
regexp = re.compile(r"(\d+)([dsmhw]?)")
220
390
234
404
return value
235
405
236
406
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()
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))
337
430
338
431
if has_actions(options) and not (options.client or options.all):
339
432
parser.error("Options require clients names or --all.")
344
437
parser.error("--dump-json can only be used alone.")
345
438
if options.all and not has_actions(options):
346
439
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
353
try:
354
bus = dbus.SystemBus()
355
mandos_dbus_objc = bus.get_object(busname, server_path)
356
except dbus.exceptions.DBusException:
357
print("Could not connect to Mandos server", file=sys.stderr)
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)
358
542
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:
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 PrintCmd(Command):
572
"""Abstract class for commands printing 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
def run(self, clients, bus=None, mandos=None):
581
print(self.output(clients.values()))
582
def output(self, clients):
583
raise NotImplementedError()
584
585
586
class DumpJSONCmd(PrintCmd):
587
def output(self, clients):
588
data = {client["Name"]:
589
{key: self.dbus_boolean_to_bool(client[key])
590
for key in self.all_keywords}
591
for client in clients.values()}
592
return json.dumps(data, indent=4, separators=(',', ': '))
593
@staticmethod
594
def dbus_boolean_to_bool(value):
595
if isinstance(value, dbus.Boolean):
596
value = bool(value)
597
return value
598
599
600
class PrintTableCmd(PrintCmd):
601
def __init__(self, verbose=False):
602
self.verbose = verbose
603
604
def output(self, clients):
605
default_keywords = ("Name", "Enabled", "Timeout",
606
"LastCheckedOK")
607
keywords = default_keywords
608
if self.verbose:
609
keywords = self.all_keywords
610
return str(self.TableOfClients(clients, keywords))
611
612
class TableOfClients(object):
613
tableheaders = {
614
"Name": "Name",
615
"Enabled": "Enabled",
616
"Timeout": "Timeout",
617
"LastCheckedOK": "Last Successful Check",
618
"LastApprovalRequest": "Last Approval Request",
619
"Created": "Created",
620
"Interval": "Interval",
621
"Host": "Host",
622
"Fingerprint": "Fingerprint",
623
"KeyID": "Key ID",
624
"CheckerRunning": "Check Is Running",
625
"LastEnabled": "Last Enabled",
626
"ApprovalPending": "Approval Is Pending",
627
"ApprovedByDefault": "Approved By Default",
628
"ApprovalDelay": "Approval Delay",
629
"ApprovalDuration": "Approval Duration",
630
"Checker": "Checker",
631
"ExtendedTimeout": "Extended Timeout",
632
"Expires": "Expires",
633
"LastCheckerStatus": "Last Checker Status",
634
}
635
636
def __init__(self, clients, keywords, tableheaders=None):
637
self.clients = clients
638
self.keywords = keywords
639
if tableheaders is not None:
640
self.tableheaders = tableheaders
641
642
def __str__(self):
643
return "\n".join(self.rows())
644
645
if sys.version_info.major == 2:
646
__unicode__ = __str__
647
def __str__(self):
648
return str(self).encode(locale.getpreferredencoding())
649
650
def rows(self):
651
format_string = self.row_formatting_string()
652
rows = [self.header_line(format_string)]
653
rows.extend(self.client_line(client, format_string)
654
for client in self.clients)
655
return rows
656
657
def row_formatting_string(self):
658
"Format string used to format table rows"
659
return " ".join("{{{key}:{width}}}".format(
660
width=max(len(self.tableheaders[key]),
661
*(len(self.string_from_client(client, key))
662
for client in self.clients)),
663
key=key)
664
for key in self.keywords)
665
666
def string_from_client(self, client, key):
667
return self.valuetostring(client[key], key)
668
669
@staticmethod
670
def valuetostring(value, keyword):
671
if isinstance(value, dbus.Boolean):
672
return "Yes" if value else "No"
673
if keyword in ("Timeout", "Interval", "ApprovalDelay",
674
"ApprovalDuration", "ExtendedTimeout"):
675
return milliseconds_to_string(value)
676
return str(value)
677
678
def header_line(self, format_string):
679
return format_string.format(**self.tableheaders)
680
681
def client_line(self, client, format_string):
682
return format_string.format(
683
**{key: self.string_from_client(client, key)
684
for key in self.keywords})
685
686
687
def milliseconds_to_string(ms):
688
td = datetime.timedelta(0, 0, 0, ms)
689
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
690
.format(days="{}T".format(td.days) if td.days else "",
691
hours=td.seconds // 3600,
692
minutes=(td.seconds % 3600) // 60,
693
seconds=td.seconds % 60))
694
695
696
class PropertyCmd(Command):
697
"""Abstract class for Actions for setting one client property"""
698
def run_on_one_client(self, client, properties):
699
"""Set the Client's D-Bus property"""
700
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
701
client.__dbus_object_path__,
702
dbus.PROPERTIES_IFACE, client_dbus_interface,
703
self.propname, self.value_to_set
704
if not isinstance(self.value_to_set, dbus.Boolean)
705
else bool(self.value_to_set))
706
client.Set(client_dbus_interface, self.propname,
707
self.value_to_set,
708
dbus_interface=dbus.PROPERTIES_IFACE)
709
@property
710
def propname(self):
711
raise NotImplementedError()
712
713
714
class EnableCmd(PropertyCmd):
715
propname = "Enabled"
716
value_to_set = dbus.Boolean(True)
717
718
719
class DisableCmd(PropertyCmd):
720
propname = "Enabled"
721
value_to_set = dbus.Boolean(False)
722
723
724
class BumpTimeoutCmd(PropertyCmd):
725
propname = "LastCheckedOK"
726
value_to_set = ""
727
728
729
class StartCheckerCmd(PropertyCmd):
730
propname = "CheckerRunning"
731
value_to_set = dbus.Boolean(True)
732
733
734
class StopCheckerCmd(PropertyCmd):
735
propname = "CheckerRunning"
736
value_to_set = dbus.Boolean(False)
737
738
739
class ApproveByDefaultCmd(PropertyCmd):
740
propname = "ApprovedByDefault"
741
value_to_set = dbus.Boolean(True)
742
743
744
class DenyByDefaultCmd(PropertyCmd):
745
propname = "ApprovedByDefault"
746
value_to_set = dbus.Boolean(False)
747
748
749
class PropertyValueCmd(PropertyCmd):
750
"""Abstract class for PropertyCmd recieving a value as argument"""
751
def __init__(self, value):
752
self.value_to_set = value
753
754
755
class SetCheckerCmd(PropertyValueCmd):
756
propname = "Checker"
757
758
759
class SetHostCmd(PropertyValueCmd):
760
propname = "Host"
761
762
763
class SetSecretCmd(PropertyValueCmd):
764
propname = "Secret"
765
@property
766
def value_to_set(self):
767
return self._vts
768
@value_to_set.setter
769
def value_to_set(self, value):
770
"""When setting, read data from supplied file object"""
771
self._vts = value.read()
772
value.close()
773
774
775
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
776
"""Abstract class for PropertyValueCmd taking a value argument as
777
a datetime.timedelta() but should store it as milliseconds."""
778
@property
779
def value_to_set(self):
780
return self._vts
781
@value_to_set.setter
782
def value_to_set(self, value):
783
"""When setting, convert value from a datetime.timedelta"""
784
self._vts = int(round(value.total_seconds() * 1000))
785
786
787
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
788
propname = "Timeout"
789
790
791
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
792
propname = "ExtendedTimeout"
793
794
795
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
796
propname = "Interval"
797
798
799
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
800
propname = "ApprovalDelay"
801
802
803
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
804
propname = "ApprovalDuration"
805
806
807
808
class Test_string_to_delta(unittest.TestCase):
809
def test_handles_basic_rfc3339(self):
810
self.assertEqual(string_to_delta("PT0S"),
811
datetime.timedelta())
812
self.assertEqual(string_to_delta("P0D"),
813
datetime.timedelta())
814
self.assertEqual(string_to_delta("PT1S"),
815
datetime.timedelta(0, 1))
816
self.assertEqual(string_to_delta("PT2H"),
817
datetime.timedelta(0, 7200))
818
def test_falls_back_to_pre_1_6_1_with_warning(self):
819
# assertLogs only exists in Python 3.4
820
if hasattr(self, "assertLogs"):
821
with self.assertLogs(log, logging.WARNING):
822
value = string_to_delta("2h")
823
else:
824
class WarningFilter(logging.Filter):
825
"""Don't show, but record the presence of, warnings"""
826
def filter(self, record):
827
is_warning = record.levelno >= logging.WARNING
828
self.found = is_warning or getattr(self, "found",
829
False)
830
return not is_warning
831
warning_filter = WarningFilter()
832
log.addFilter(warning_filter)
833
try:
834
value = string_to_delta("2h")
835
finally:
836
log.removeFilter(warning_filter)
837
self.assertTrue(getattr(warning_filter, "found", False))
838
self.assertEqual(value, datetime.timedelta(0, 7200))
839
840
841
class Test_check_option_syntax(unittest.TestCase):
842
# This mostly corresponds to the definition from has_actions() in
843
# check_option_syntax()
844
actions = {
845
# The actual values set here are not that important, but we do
846
# at least stick to the correct types, even though they are
847
# never used
848
"enable": True,
849
"disable": True,
850
"bump_timeout": True,
851
"start_checker": True,
852
"stop_checker": True,
853
"is_enabled": True,
854
"remove": True,
855
"checker": "x",
856
"timeout": datetime.timedelta(),
857
"extended_timeout": datetime.timedelta(),
858
"interval": datetime.timedelta(),
859
"approved_by_default": True,
860
"approval_delay": datetime.timedelta(),
861
"approval_duration": datetime.timedelta(),
862
"host": "x",
863
"secret": io.BytesIO(b"x"),
864
"approve": True,
865
"deny": True,
866
}
867
868
def setUp(self):
869
self.parser = argparse.ArgumentParser()
870
add_command_line_options(self.parser)
871
872
@contextlib.contextmanager
873
def assertParseError(self):
874
with self.assertRaises(SystemExit) as e:
875
with self.temporarily_suppress_stderr():
876
yield
877
# Exit code from argparse is guaranteed to be "2". Reference:
878
# https://docs.python.org/3/library
879
# /argparse.html#exiting-methods
880
self.assertEqual(e.exception.code, 2)
881
882
@staticmethod
883
@contextlib.contextmanager
884
def temporarily_suppress_stderr():
885
null = os.open(os.path.devnull, os.O_RDWR)
886
stderrcopy = os.dup(sys.stderr.fileno())
887
os.dup2(null, sys.stderr.fileno())
888
os.close(null)
371
889
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}
890
yield
377
891
finally:
378
892
# restore stderr
379
893
os.dup2(stderrcopy, sys.stderr.fileno())
380
894
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
895
896
def check_option_syntax(self, options):
897
check_option_syntax(self.parser, options)
898
899
def test_actions_requires_client_or_all(self):
900
for action, value in self.actions.items():
901
options = self.parser.parse_args()
902
setattr(options, action, value)
903
with self.assertParseError():
904
self.check_option_syntax(options)
905
906
def test_actions_conflicts_with_verbose(self):
907
for action, value in self.actions.items():
908
options = self.parser.parse_args()
909
setattr(options, action, value)
910
options.verbose = True
911
with self.assertParseError():
912
self.check_option_syntax(options)
913
914
def test_dump_json_conflicts_with_verbose(self):
915
options = self.parser.parse_args()
916
options.dump_json = True
917
options.verbose = True
918
with self.assertParseError():
919
self.check_option_syntax(options)
920
921
def test_dump_json_conflicts_with_action(self):
922
for action, value in self.actions.items():
923
options = self.parser.parse_args()
924
setattr(options, action, value)
925
options.dump_json = True
926
with self.assertParseError():
927
self.check_option_syntax(options)
928
929
def test_all_can_not_be_alone(self):
930
options = self.parser.parse_args()
931
options.all = True
932
with self.assertParseError():
933
self.check_option_syntax(options)
934
935
def test_all_is_ok_with_any_action(self):
936
for action, value in self.actions.items():
937
options = self.parser.parse_args()
938
setattr(options, action, value)
939
options.all = True
940
self.check_option_syntax(options)
941
942
def test_is_enabled_fails_without_client(self):
943
options = self.parser.parse_args()
944
options.is_enabled = True
945
with self.assertParseError():
946
self.check_option_syntax(options)
947
948
def test_is_enabled_works_with_one_client(self):
949
options = self.parser.parse_args()
950
options.is_enabled = True
951
options.client = ["foo"]
952
self.check_option_syntax(options)
953
954
def test_is_enabled_fails_with_two_clients(self):
955
options = self.parser.parse_args()
956
options.is_enabled = True
957
options.client = ["foo", "barbar"]
958
with self.assertParseError():
959
self.check_option_syntax(options)
960
961
def test_remove_can_only_be_combined_with_action_deny(self):
962
for action, value in self.actions.items():
963
if action in {"remove", "deny"}:
964
continue
965
options = self.parser.parse_args()
966
setattr(options, action, value)
967
options.all = True
968
options.remove = True
969
with self.assertParseError():
970
self.check_option_syntax(options)
971
972
973
class Test_command_from_options(unittest.TestCase):
974
def setUp(self):
975
self.parser = argparse.ArgumentParser()
976
add_command_line_options(self.parser)
977
def assert_command_from_args(self, args, command_cls,
978
**cmd_attrs):
979
"""Assert that parsing ARGS should result in an instance of
980
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
981
options = self.parser.parse_args(args)
982
check_option_syntax(self.parser, options)
983
commands = commands_from_options(options)
984
self.assertEqual(len(commands), 1)
985
command = commands[0]
986
self.assertIsInstance(command, command_cls)
987
for key, value in cmd_attrs.items():
988
self.assertEqual(getattr(command, key), value)
989
def test_print_table(self):
990
self.assert_command_from_args([], PrintTableCmd,
991
verbose=False)
992
993
def test_print_table_verbose(self):
994
self.assert_command_from_args(["--verbose"], PrintTableCmd,
995
verbose=True)
996
997
def test_print_table_verbose_short(self):
998
self.assert_command_from_args(["-v"], PrintTableCmd,
999
verbose=True)
1000
1001
def test_enable(self):
1002
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1003
1004
def test_enable_short(self):
1005
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1006
1007
def test_disable(self):
1008
self.assert_command_from_args(["--disable", "foo"],
1009
DisableCmd)
1010
1011
def test_disable_short(self):
1012
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1013
1014
def test_bump_timeout(self):
1015
self.assert_command_from_args(["--bump-timeout", "foo"],
1016
BumpTimeoutCmd)
1017
1018
def test_bump_timeout_short(self):
1019
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1020
1021
def test_start_checker(self):
1022
self.assert_command_from_args(["--start-checker", "foo"],
1023
StartCheckerCmd)
1024
1025
def test_stop_checker(self):
1026
self.assert_command_from_args(["--stop-checker", "foo"],
1027
StopCheckerCmd)
1028
1029
def test_remove(self):
1030
self.assert_command_from_args(["--remove", "foo"],
1031
RemoveCmd)
1032
1033
def test_remove_short(self):
1034
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1035
1036
def test_checker(self):
1037
self.assert_command_from_args(["--checker", ":", "foo"],
1038
SetCheckerCmd, value_to_set=":")
1039
1040
def test_checker_empty(self):
1041
self.assert_command_from_args(["--checker", "", "foo"],
1042
SetCheckerCmd, value_to_set="")
1043
1044
def test_checker_short(self):
1045
self.assert_command_from_args(["-c", ":", "foo"],
1046
SetCheckerCmd, value_to_set=":")
1047
1048
def test_timeout(self):
1049
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1050
SetTimeoutCmd,
1051
value_to_set=300000)
1052
1053
def test_timeout_short(self):
1054
self.assert_command_from_args(["-t", "PT5M", "foo"],
1055
SetTimeoutCmd,
1056
value_to_set=300000)
1057
1058
def test_extended_timeout(self):
1059
self.assert_command_from_args(["--extended-timeout", "PT15M",
1060
"foo"],
1061
SetExtendedTimeoutCmd,
1062
value_to_set=900000)
1063
1064
def test_interval(self):
1065
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1066
SetIntervalCmd,
1067
value_to_set=120000)
1068
1069
def test_interval_short(self):
1070
self.assert_command_from_args(["-i", "PT2M", "foo"],
1071
SetIntervalCmd,
1072
value_to_set=120000)
1073
1074
def test_approve_by_default(self):
1075
self.assert_command_from_args(["--approve-by-default", "foo"],
1076
ApproveByDefaultCmd)
1077
1078
def test_deny_by_default(self):
1079
self.assert_command_from_args(["--deny-by-default", "foo"],
1080
DenyByDefaultCmd)
1081
1082
def test_approval_delay(self):
1083
self.assert_command_from_args(["--approval-delay", "PT30S",
1084
"foo"], SetApprovalDelayCmd,
1085
value_to_set=30000)
1086
1087
def test_approval_duration(self):
1088
self.assert_command_from_args(["--approval-duration", "PT1S",
1089
"foo"], SetApprovalDurationCmd,
1090
value_to_set=1000)
1091
1092
def test_host(self):
1093
self.assert_command_from_args(["--host", "foo.example.org",
1094
"foo"], SetHostCmd,
1095
value_to_set="foo.example.org")
1096
1097
def test_host_short(self):
1098
self.assert_command_from_args(["-H", "foo.example.org",
1099
"foo"], SetHostCmd,
1100
value_to_set="foo.example.org")
1101
1102
def test_secret_devnull(self):
1103
self.assert_command_from_args(["--secret", os.path.devnull,
1104
"foo"], SetSecretCmd,
1105
value_to_set=b"")
1106
1107
def test_secret_tempfile(self):
1108
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1109
value = b"secret\0xyzzy\nbar"
1110
f.write(value)
1111
f.seek(0)
1112
self.assert_command_from_args(["--secret", f.name,
1113
"foo"], SetSecretCmd,
1114
value_to_set=value)
1115
1116
def test_secret_devnull_short(self):
1117
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1118
SetSecretCmd, value_to_set=b"")
1119
1120
def test_secret_tempfile_short(self):
1121
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1122
value = b"secret\0xyzzy\nbar"
1123
f.write(value)
1124
f.seek(0)
1125
self.assert_command_from_args(["-s", f.name, "foo"],
1126
SetSecretCmd,
1127
value_to_set=value)
1128
1129
def test_approve(self):
1130
self.assert_command_from_args(["--approve", "foo"],
1131
ApproveCmd)
1132
1133
def test_approve_short(self):
1134
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1135
1136
def test_deny(self):
1137
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1138
1139
def test_deny_short(self):
1140
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1141
1142
def test_dump_json(self):
1143
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1144
1145
def test_is_enabled(self):
1146
self.assert_command_from_args(["--is-enabled", "foo"],
1147
IsEnabledCmd)
1148
1149
def test_is_enabled_short(self):
1150
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1151
1152
def test_deny_before_remove(self):
1153
options = self.parser.parse_args(["--deny", "--remove",
1154
"foo"])
1155
check_option_syntax(self.parser, options)
1156
commands = commands_from_options(options)
1157
self.assertEqual(len(commands), 2)
1158
self.assertIsInstance(commands[0], DenyCmd)
1159
self.assertIsInstance(commands[1], RemoveCmd)
1160
1161
def test_deny_before_remove_reversed(self):
1162
options = self.parser.parse_args(["--remove", "--deny",
1163
"--all"])
1164
check_option_syntax(self.parser, options)
1165
commands = commands_from_options(options)
1166
self.assertEqual(len(commands), 2)
1167
self.assertIsInstance(commands[0], DenyCmd)
1168
self.assertIsInstance(commands[1], RemoveCmd)
1169
1170
1171
class TestCmd(unittest.TestCase):
1172
"""Abstract class for tests of command classes"""
1173
def setUp(self):
1174
testcase = self
1175
class MockClient(object):
1176
def __init__(self, name, **attributes):
1177
self.__dbus_object_path__ = "/clients/{}".format(name)
1178
self.attributes = attributes
1179
self.attributes["Name"] = name
1180
self.calls = []
1181
def Set(self, interface, propname, value, dbus_interface):
1182
testcase.assertEqual(interface, client_dbus_interface)
1183
testcase.assertEqual(dbus_interface,
1184
dbus.PROPERTIES_IFACE)
1185
self.attributes[propname] = value
1186
def Get(self, interface, propname, dbus_interface):
1187
testcase.assertEqual(interface, client_dbus_interface)
1188
testcase.assertEqual(dbus_interface,
1189
dbus.PROPERTIES_IFACE)
1190
return self.attributes[propname]
1191
def Approve(self, approve, dbus_interface):
1192
testcase.assertEqual(dbus_interface,
1193
client_dbus_interface)
1194
self.calls.append(("Approve", (approve,
1195
dbus_interface)))
1196
self.client = MockClient(
1197
"foo",
1198
KeyID=("92ed150794387c03ce684574b1139a65"
1199
"94a34f895daaaf09fd8ea90a27cddb12"),
1200
Secret=b"secret",
1201
Host="foo.example.org",
1202
Enabled=dbus.Boolean(True),
1203
Timeout=300000,
1204
LastCheckedOK="2019-02-03T00:00:00",
1205
Created="2019-01-02T00:00:00",
1206
Interval=120000,
1207
Fingerprint=("778827225BA7DE539C5A"
1208
"7CFA59CFF7CDBD9A5920"),
1209
CheckerRunning=dbus.Boolean(False),
1210
LastEnabled="2019-01-03T00:00:00",
1211
ApprovalPending=dbus.Boolean(False),
1212
ApprovedByDefault=dbus.Boolean(True),
1213
LastApprovalRequest="",
1214
ApprovalDelay=0,
1215
ApprovalDuration=1000,
1216
Checker="fping -q -- %(host)s",
1217
ExtendedTimeout=900000,
1218
Expires="2019-02-04T00:00:00",
1219
LastCheckerStatus=0)
1220
self.other_client = MockClient(
1221
"barbar",
1222
KeyID=("0558568eedd67d622f5c83b35a115f79"
1223
"6ab612cff5ad227247e46c2b020f441c"),
1224
Secret=b"secretbar",
1225
Host="192.0.2.3",
1226
Enabled=dbus.Boolean(True),
1227
Timeout=300000,
1228
LastCheckedOK="2019-02-04T00:00:00",
1229
Created="2019-01-03T00:00:00",
1230
Interval=120000,
1231
Fingerprint=("3E393AEAEFB84C7E89E2"
1232
"F547B3A107558FCA3A27"),
1233
CheckerRunning=dbus.Boolean(True),
1234
LastEnabled="2019-01-04T00:00:00",
1235
ApprovalPending=dbus.Boolean(False),
1236
ApprovedByDefault=dbus.Boolean(False),
1237
LastApprovalRequest="2019-01-03T00:00:00",
1238
ApprovalDelay=30000,
1239
ApprovalDuration=1000,
1240
Checker=":",
1241
ExtendedTimeout=900000,
1242
Expires="2019-02-05T00:00:00",
1243
LastCheckerStatus=-2)
1244
self.clients = collections.OrderedDict(
1245
[
1246
("/clients/foo", self.client.attributes),
1247
("/clients/barbar", self.other_client.attributes),
1248
])
1249
self.one_client = {"/clients/foo": self.client.attributes}
1250
@property
1251
def bus(self):
1252
class Bus(object):
1253
@staticmethod
1254
def get_object(client_bus_name, path):
1255
self.assertEqual(client_bus_name, dbus_busname)
1256
return {
1257
"/clients/foo": self.client,
1258
"/clients/barbar": self.other_client,
1259
}[path]
1260
return Bus()
1261
1262
1263
class TestIsEnabledCmd(TestCmd):
1264
def test_is_enabled(self):
1265
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1266
properties)
1267
for client, properties
1268
in self.clients.items()))
1269
def test_is_enabled_run_exits_successfully(self):
1270
with self.assertRaises(SystemExit) as e:
1271
IsEnabledCmd().run(self.one_client)
1272
if e.exception.code is not None:
1273
self.assertEqual(e.exception.code, 0)
1274
else:
1275
self.assertIsNone(e.exception.code)
1276
def test_is_enabled_run_exits_with_failure(self):
1277
self.client.attributes["Enabled"] = dbus.Boolean(False)
1278
with self.assertRaises(SystemExit) as e:
1279
IsEnabledCmd().run(self.one_client)
1280
if isinstance(e.exception.code, int):
1281
self.assertNotEqual(e.exception.code, 0)
1282
else:
1283
self.assertIsNotNone(e.exception.code)
1284
1285
1286
class TestApproveCmd(TestCmd):
1287
def test_approve(self):
1288
ApproveCmd().run(self.clients, self.bus)
1289
for clientpath in self.clients:
1290
client = self.bus.get_object(dbus_busname, clientpath)
1291
self.assertIn(("Approve", (True, client_dbus_interface)),
1292
client.calls)
1293
1294
1295
class TestDenyCmd(TestCmd):
1296
def test_deny(self):
1297
DenyCmd().run(self.clients, self.bus)
1298
for clientpath in self.clients:
1299
client = self.bus.get_object(dbus_busname, clientpath)
1300
self.assertIn(("Approve", (False, client_dbus_interface)),
1301
client.calls)
1302
1303
class TestRemoveCmd(TestCmd):
1304
def test_remove(self):
1305
class MockMandos(object):
1306
def __init__(self):
1307
self.calls = []
1308
def RemoveClient(self, dbus_path):
1309
self.calls.append(("RemoveClient", (dbus_path,)))
1310
mandos = MockMandos()
1311
super(TestRemoveCmd, self).setUp()
1312
RemoveCmd().run(self.clients, self.bus, mandos)
1313
self.assertEqual(len(mandos.calls), 2)
1314
for clientpath in self.clients:
1315
self.assertIn(("RemoveClient", (clientpath,)),
1316
mandos.calls)
1317
1318
1319
class TestDumpJSONCmd(TestCmd):
1320
def setUp(self):
1321
self.expected_json = {
1322
"foo": {
1323
"Name": "foo",
1324
"KeyID": ("92ed150794387c03ce684574b1139a65"
1325
"94a34f895daaaf09fd8ea90a27cddb12"),
1326
"Host": "foo.example.org",
1327
"Enabled": True,
1328
"Timeout": 300000,
1329
"LastCheckedOK": "2019-02-03T00:00:00",
1330
"Created": "2019-01-02T00:00:00",
1331
"Interval": 120000,
1332
"Fingerprint": ("778827225BA7DE539C5A"
1333
"7CFA59CFF7CDBD9A5920"),
1334
"CheckerRunning": False,
1335
"LastEnabled": "2019-01-03T00:00:00",
1336
"ApprovalPending": False,
1337
"ApprovedByDefault": True,
1338
"LastApprovalRequest": "",
1339
"ApprovalDelay": 0,
1340
"ApprovalDuration": 1000,
1341
"Checker": "fping -q -- %(host)s",
1342
"ExtendedTimeout": 900000,
1343
"Expires": "2019-02-04T00:00:00",
1344
"LastCheckerStatus": 0,
1345
},
1346
"barbar": {
1347
"Name": "barbar",
1348
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1349
"6ab612cff5ad227247e46c2b020f441c"),
1350
"Host": "192.0.2.3",
1351
"Enabled": True,
1352
"Timeout": 300000,
1353
"LastCheckedOK": "2019-02-04T00:00:00",
1354
"Created": "2019-01-03T00:00:00",
1355
"Interval": 120000,
1356
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1357
"F547B3A107558FCA3A27"),
1358
"CheckerRunning": True,
1359
"LastEnabled": "2019-01-04T00:00:00",
1360
"ApprovalPending": False,
1361
"ApprovedByDefault": False,
1362
"LastApprovalRequest": "2019-01-03T00:00:00",
1363
"ApprovalDelay": 30000,
1364
"ApprovalDuration": 1000,
1365
"Checker": ":",
1366
"ExtendedTimeout": 900000,
1367
"Expires": "2019-02-05T00:00:00",
1368
"LastCheckerStatus": -2,
1369
},
1370
}
1371
return super(TestDumpJSONCmd, self).setUp()
1372
def test_normal(self):
1373
json_data = json.loads(DumpJSONCmd().output(self.clients))
1374
self.assertDictEqual(json_data, self.expected_json)
1375
def test_one_client(self):
1376
clients = self.one_client
1377
json_data = json.loads(DumpJSONCmd().output(clients))
1378
expected_json = {"foo": self.expected_json["foo"]}
1379
self.assertDictEqual(json_data, expected_json)
1380
1381
1382
class TestPrintTableCmd(TestCmd):
1383
def test_normal(self):
1384
output = PrintTableCmd().output(self.clients.values())
1385
expected_output = "\n".join((
1386
"Name Enabled Timeout Last Successful Check",
1387
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1388
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1389
))
1390
self.assertEqual(output, expected_output)
1391
def test_verbose(self):
1392
output = PrintTableCmd(verbose=True).output(
1393
self.clients.values())
1394
columns = (
1395
(
1396
"Name ",
1397
"foo ",
1398
"barbar ",
1399
),(
1400
"Enabled ",
1401
"Yes ",
1402
"Yes ",
1403
),(
1404
"Timeout ",
1405
"00:05:00 ",
1406
"00:05:00 ",
1407
),(
1408
"Last Successful Check ",
1409
"2019-02-03T00:00:00 ",
1410
"2019-02-04T00:00:00 ",
1411
),(
1412
"Created ",
1413
"2019-01-02T00:00:00 ",
1414
"2019-01-03T00:00:00 ",
1415
),(
1416
"Interval ",
1417
"00:02:00 ",
1418
"00:02:00 ",
1419
),(
1420
"Host ",
1421
"foo.example.org ",
1422
"192.0.2.3 ",
1423
),(
1424
("Key ID "
1425
" "),
1426
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1427
"ea90a27cddb12 "),
1428
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1429
"46c2b020f441c "),
1430
),(
1431
"Fingerprint ",
1432
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1433
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1434
),(
1435
"Check Is Running ",
1436
"No ",
1437
"Yes ",
1438
),(
1439
"Last Enabled ",
1440
"2019-01-03T00:00:00 ",
1441
"2019-01-04T00:00:00 ",
1442
),(
1443
"Approval Is Pending ",
1444
"No ",
1445
"No ",
1446
),(
1447
"Approved By Default ",
1448
"Yes ",
1449
"No ",
1450
),(
1451
"Last Approval Request ",
1452
" ",
1453
"2019-01-03T00:00:00 ",
1454
),(
1455
"Approval Delay ",
1456
"00:00:00 ",
1457
"00:00:30 ",
1458
),(
1459
"Approval Duration ",
1460
"00:00:01 ",
1461
"00:00:01 ",
1462
),(
1463
"Checker ",
1464
"fping -q -- %(host)s ",
1465
": ",
1466
),(
1467
"Extended Timeout ",
1468
"00:15:00 ",
1469
"00:15:00 ",
1470
),(
1471
"Expires ",
1472
"2019-02-04T00:00:00 ",
1473
"2019-02-05T00:00:00 ",
1474
),(
1475
"Last Checker Status",
1476
"0 ",
1477
"-2 ",
1478
)
1479
)
1480
num_lines = max(len(rows) for rows in columns)
1481
expected_output = "\n".join("".join(rows[line]
1482
for rows in columns)
1483
for line in range(num_lines))
1484
self.assertEqual(output, expected_output)
1485
def test_one_client(self):
1486
output = PrintTableCmd().output(self.one_client.values())
1487
expected_output = """
1488
Name Enabled Timeout Last Successful Check
1489
foo Yes 00:05:00 2019-02-03T00:00:00
1490
"""[1:-1]
1491
self.assertEqual(output, expected_output)
1492
1493
1494
class Test_milliseconds_to_string(unittest.TestCase):
1495
def test_all(self):
1496
self.assertEqual(milliseconds_to_string(93785000),
1497
"1T02:03:05")
1498
def test_no_days(self):
1499
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
1500
def test_all_zero(self):
1501
self.assertEqual(milliseconds_to_string(0), "00:00:00")
1502
def test_no_fractional_seconds(self):
1503
self.assertEqual(milliseconds_to_string(400), "00:00:00")
1504
self.assertEqual(milliseconds_to_string(900), "00:00:00")
1505
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
1506
1507
1508
class Unique(object):
1509
"""Class for objects which exist only to be unique objects, since
1510
unittest.mock.sentinel only exists in Python 3.3"""
1511
1512
1513
class TestPropertyCmd(TestCmd):
1514
"""Abstract class for tests of PropertyCmd classes"""
1515
def runTest(self):
1516
if not hasattr(self, "command"):
1517
return
1518
values_to_get = getattr(self, "values_to_get",
1519
self.values_to_set)
1520
for value_to_set, value_to_get in zip(self.values_to_set,
1521
values_to_get):
1522
for clientpath in self.clients:
1523
client = self.bus.get_object(dbus_busname, clientpath)
1524
old_value = client.attributes[self.propname]
1525
self.assertNotIsInstance(old_value, Unique)
1526
client.attributes[self.propname] = Unique()
1527
self.run_command(value_to_set, self.clients)
1528
for clientpath in self.clients:
1529
client = self.bus.get_object(dbus_busname, clientpath)
1530
value = client.attributes[self.propname]
1531
self.assertNotIsInstance(value, Unique)
1532
self.assertEqual(value, value_to_get)
1533
def run_command(self, value, clients):
1534
self.command().run(clients, self.bus)
1535
1536
1537
class TestEnableCmd(TestCmd):
1538
def test_enable(self):
1539
for clientpath in self.clients:
1540
client = self.bus.get_object(dbus_busname, clientpath)
1541
client.attributes["Enabled"] = False
1542
1543
EnableCmd().run(self.clients, self.bus)
1544
1545
for clientpath in self.clients:
1546
client = self.bus.get_object(dbus_busname, clientpath)
1547
self.assertTrue(client.attributes["Enabled"])
1548
1549
1550
class TestDisableCmd(TestCmd):
1551
def test_disable(self):
1552
DisableCmd().run(self.clients, self.bus)
1553
for clientpath in self.clients:
1554
client = self.bus.get_object(dbus_busname, clientpath)
1555
self.assertFalse(client.attributes["Enabled"])
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
def runTest(self):
1591
if type(self) is TestPropertyValueCmd:
1592
return
1593
return super(TestPropertyValueCmd, self).runTest()
1594
def run_command(self, value, clients):
1595
self.command(value).run(clients, self.bus)
1596
1597
1598
class TestSetCheckerCmd(TestPropertyValueCmd):
1599
command = SetCheckerCmd
1600
propname = "Checker"
1601
values_to_set = ["", ":", "fping -q -- %s"]
1602
1603
1604
class TestSetHostCmd(TestPropertyValueCmd):
1605
command = SetHostCmd
1606
propname = "Host"
1607
values_to_set = ["192.0.2.3", "foo.example.org"]
1608
1609
1610
class TestSetSecretCmd(TestPropertyValueCmd):
1611
command = SetSecretCmd
1612
propname = "Secret"
1613
values_to_set = [io.BytesIO(b""),
1614
io.BytesIO(b"secret\0xyzzy\nbar")]
1615
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1616
1617
1618
class TestSetTimeoutCmd(TestPropertyValueCmd):
1619
command = SetTimeoutCmd
1620
propname = "Timeout"
1621
values_to_set = [datetime.timedelta(),
1622
datetime.timedelta(minutes=5),
1623
datetime.timedelta(seconds=1),
1624
datetime.timedelta(weeks=1),
1625
datetime.timedelta(weeks=52)]
1626
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1627
1628
1629
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1630
command = SetExtendedTimeoutCmd
1631
propname = "ExtendedTimeout"
1632
values_to_set = [datetime.timedelta(),
1633
datetime.timedelta(minutes=5),
1634
datetime.timedelta(seconds=1),
1635
datetime.timedelta(weeks=1),
1636
datetime.timedelta(weeks=52)]
1637
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1638
1639
1640
class TestSetIntervalCmd(TestPropertyValueCmd):
1641
command = SetIntervalCmd
1642
propname = "Interval"
1643
values_to_set = [datetime.timedelta(),
1644
datetime.timedelta(minutes=5),
1645
datetime.timedelta(seconds=1),
1646
datetime.timedelta(weeks=1),
1647
datetime.timedelta(weeks=52)]
1648
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1649
1650
1651
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1652
command = SetApprovalDelayCmd
1653
propname = "ApprovalDelay"
1654
values_to_set = [datetime.timedelta(),
1655
datetime.timedelta(minutes=5),
1656
datetime.timedelta(seconds=1),
1657
datetime.timedelta(weeks=1),
1658
datetime.timedelta(weeks=52)]
1659
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1660
1661
1662
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1663
command = SetApprovalDurationCmd
1664
propname = "ApprovalDuration"
1665
values_to_set = [datetime.timedelta(),
1666
datetime.timedelta(minutes=5),
1667
datetime.timedelta(seconds=1),
1668
datetime.timedelta(weeks=1),
1669
datetime.timedelta(weeks=52)]
1670
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1671
1672
1673
1674
def should_only_run_tests():
1675
parser = argparse.ArgumentParser(add_help=False)
1676
parser.add_argument("--check", action='store_true')
1677
args, unknown_args = parser.parse_known_args()
1678
run_tests = args.check
1679
if run_tests:
1680
# Remove --check argument from sys.argv
1681
sys.argv[1:] = unknown_args
1682
return run_tests
1683
1684
# Add all tests from doctest strings
1685
def load_tests(loader, tests, none):
1686
import doctest
1687
tests.addTests(doctest.DocTestSuite())
1688
return tests
496
1689
497
1690
if __name__ == "__main__":
498
main()
1691
if should_only_run_tests():
1692
# Call using ./tdd-python-script --check [--verbose]
1693
unittest.main()
1694
else:
1695
main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0