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.
- files added:
- debian/mandos-client.templates
- files removed:
- initramfs-tools-hook-conf
- 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
"CheckerRunning": "Check Is Running",
62
"LastEnabled": "Last Enabled",
63
"ApprovalPending": "Approval Is Pending",
64
"ApprovedByDefault": "Approved By Default",
65
"ApprovalDelay": "Approval Delay",
66
"ApprovalDuration": "Approval Duration",
67
"Checker": "Checker",
68
"ExtendedTimeout": "Extended Timeout",
69
"Expires": "Expires",
70
"LastCheckerStatus": "Last Checker Status",
71
}
72
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
73
domain = "se.recompile"
74
busname = domain + ".Mandos"
75
server_path = "/"
76
server_interface = domain + ".Mandos"
77
client_interface = domain + ".Mandos.Client"
78
version = "1.7.18"
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"
79
74
80
75
81
76
try:
84
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
85
80
86
81
87
def milliseconds_to_string(ms):
88
td = datetime.timedelta(0, 0, 0, ms)
89
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
90
.format(days="{}T".format(td.days) if td.days else "",
91
hours=td.seconds // 3600,
92
minutes=(td.seconds % 3600) // 60,
93
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)
94
232
95
233
96
234
def rfc3339_duration_to_delta(duration):
102
240
datetime.timedelta(0, 60)
103
241
>>> rfc3339_duration_to_delta("PT60M")
104
242
datetime.timedelta(0, 3600)
243
>>> rfc3339_duration_to_delta("P60M")
244
datetime.timedelta(1680)
105
245
>>> rfc3339_duration_to_delta("PT24H")
106
246
datetime.timedelta(1)
107
247
>>> rfc3339_duration_to_delta("P1W")
110
250
datetime.timedelta(0, 330)
111
251
>>> rfc3339_duration_to_delta("P1DT3M20S")
112
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'
113
282
"""
114
283
115
284
# Parsing an RFC 3339 duration with regular expressions is not
192
361
return value
193
362
194
363
195
def string_to_delta(interval):
196
"""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
197
367
198
>>> string_to_delta('7d')
368
>>> parse_pre_1_6_1_interval('7d')
199
369
datetime.timedelta(7)
200
>>> string_to_delta('60s')
370
>>> parse_pre_1_6_1_interval('60s')
201
371
datetime.timedelta(0, 60)
202
>>> string_to_delta('60m')
372
>>> parse_pre_1_6_1_interval('60m')
203
373
datetime.timedelta(0, 3600)
204
>>> string_to_delta('24h')
374
>>> parse_pre_1_6_1_interval('24h')
205
375
datetime.timedelta(1)
206
>>> string_to_delta('1w')
376
>>> parse_pre_1_6_1_interval('1w')
207
377
datetime.timedelta(7)
208
>>> string_to_delta('5m 30s')
378
>>> parse_pre_1_6_1_interval('5m 30s')
209
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
210
386
"""
211
387
212
try:
213
return rfc3339_duration_to_delta(interval)
214
except ValueError:
215
pass
216
217
388
value = datetime.timedelta(0)
218
389
regexp = re.compile(r"(\d+)([dsmhw]?)")
219
390
233
404
return value
234
405
235
406
236
def print_clients(clients, keywords):
237
def valuetostring(value, keyword):
238
if type(value) is dbus.Boolean:
239
return "Yes" if value else "No"
240
if keyword in ("Timeout", "Interval", "ApprovalDelay",
241
"ApprovalDuration", "ExtendedTimeout"):
242
return milliseconds_to_string(value)
243
return str(value)
244
245
# Create format string to print table rows
246
format_string = " ".join("{{{key}:{width}}}".format(
247
width=max(len(tablewords[key]),
248
max(len(valuetostring(client[key], key))
249
for client in clients)),
250
key=key)
251
for key in keywords)
252
# Print header line
253
print(format_string.format(**tablewords))
254
for client in clients:
255
print(format_string
256
.format(**{key: valuetostring(client[key], key)
257
for key in keywords}))
258
259
260
def has_actions(options):
261
return any((options.enable,
262
options.disable,
263
options.bump_timeout,
264
options.start_checker,
265
options.stop_checker,
266
options.is_enabled,
267
options.remove,
268
options.checker is not None,
269
options.timeout is not None,
270
options.extended_timeout is not None,
271
options.interval is not None,
272
options.approved_by_default is not None,
273
options.approval_delay is not None,
274
options.approval_duration is not None,
275
options.host is not None,
276
options.secret is not None,
277
options.approve,
278
options.deny))
279
280
281
def main():
282
parser = argparse.ArgumentParser()
283
parser.add_argument("--version", action="version",
284
version="%(prog)s {}".format(version),
285
help="show version number and exit")
286
parser.add_argument("-a", "--all", action="store_true",
287
help="Select all clients")
288
parser.add_argument("-v", "--verbose", action="store_true",
289
help="Print all fields")
290
parser.add_argument("-j", "--dump-json", action="store_true",
291
help="Dump client data in JSON format")
292
parser.add_argument("-e", "--enable", action="store_true",
293
help="Enable client")
294
parser.add_argument("-d", "--disable", action="store_true",
295
help="disable client")
296
parser.add_argument("-b", "--bump-timeout", action="store_true",
297
help="Bump timeout for client")
298
parser.add_argument("--start-checker", action="store_true",
299
help="Start checker for client")
300
parser.add_argument("--stop-checker", action="store_true",
301
help="Stop checker for client")
302
parser.add_argument("-V", "--is-enabled", action="store_true",
303
help="Check if client is enabled")
304
parser.add_argument("-r", "--remove", action="store_true",
305
help="Remove client")
306
parser.add_argument("-c", "--checker",
307
help="Set checker command for client")
308
parser.add_argument("-t", "--timeout",
309
help="Set timeout for client")
310
parser.add_argument("--extended-timeout",
311
help="Set extended timeout for client")
312
parser.add_argument("-i", "--interval",
313
help="Set checker interval for client")
314
parser.add_argument("--approve-by-default", action="store_true",
315
default=None, dest="approved_by_default",
316
help="Set client to be approved by default")
317
parser.add_argument("--deny-by-default", action="store_false",
318
dest="approved_by_default",
319
help="Set client to be denied by default")
320
parser.add_argument("--approval-delay",
321
help="Set delay before client approve/deny")
322
parser.add_argument("--approval-duration",
323
help="Set duration of one client approval")
324
parser.add_argument("-H", "--host", help="Set host for client")
325
parser.add_argument("-s", "--secret",
326
type=argparse.FileType(mode="rb"),
327
help="Set password blob (file) for client")
328
parser.add_argument("-A", "--approve", action="store_true",
329
help="Approve any current client request")
330
parser.add_argument("-D", "--deny", action="store_true",
331
help="Deny any current client request")
332
parser.add_argument("--check", action="store_true",
333
help="Run self-test")
334
parser.add_argument("client", nargs="*", help="Client name")
335
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))
336
430
337
431
if has_actions(options) and not (options.client or options.all):
338
432
parser.error("Options require clients names or --all.")
343
437
parser.error("--dump-json can only be used alone.")
344
438
if options.all and not has_actions(options):
345
439
parser.error("--all requires an action.")
346
347
if options.check:
348
import doctest
349
fail_count, test_count = doctest.testmod()
350
sys.exit(os.EX_OK if fail_count == 0 else 1)
351
352
try:
353
bus = dbus.SystemBus()
354
mandos_dbus_objc = bus.get_object(busname, server_path)
355
except dbus.exceptions.DBusException:
356
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)
357
542
sys.exit(1)
358
359
mandos_serv = dbus.Interface(mandos_dbus_objc,
360
dbus_interface=server_interface)
361
mandos_serv_object_manager = dbus.Interface(
362
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
363
364
# block stderr since dbus library prints to stderr
365
null = os.open(os.path.devnull, os.O_RDWR)
366
stderrcopy = os.dup(sys.stderr.fileno())
367
os.dup2(null, sys.stderr.fileno())
368
os.close(null)
369
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)
370
889
try:
371
mandos_clients = {path: ifs_and_props[client_interface]
372
for path, ifs_and_props in
373
mandos_serv_object_manager
374
.GetManagedObjects().items()
375
if client_interface in ifs_and_props}
890
yield
376
891
finally:
377
892
# restore stderr
378
893
os.dup2(stderrcopy, sys.stderr.fileno())
379
894
os.close(stderrcopy)
380
except dbus.exceptions.DBusException as e:
381
print("Access denied: "
382
"Accessing mandos server through D-Bus: {}".format(e),
383
file=sys.stderr)
384
sys.exit(1)
385
386
# Compile dict of (clients: properties) to process
387
clients = {}
388
389
if options.all or not options.client:
390
clients = {bus.get_object(busname, path): properties
391
for path, properties in mandos_clients.items()}
392
else:
393
for name in options.client:
394
for path, client in mandos_clients.items():
395
if client["Name"] == name:
396
client_objc = bus.get_object(busname, path)
397
clients[client_objc] = client
398
break
399
else:
400
print("Client not found on server: {!r}"
401
.format(name), file=sys.stderr)
402
sys.exit(1)
403
404
if not has_actions(options) and clients:
405
if options.verbose or options.dump_json:
406
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
407
"Created", "Interval", "Host", "Fingerprint",
408
"CheckerRunning", "LastEnabled",
409
"ApprovalPending", "ApprovedByDefault",
410
"LastApprovalRequest", "ApprovalDelay",
411
"ApprovalDuration", "Checker",
412
"ExtendedTimeout", "Expires",
413
"LastCheckerStatus")
414
else:
415
keywords = defaultkeywords
416
417
if options.dump_json:
418
json.dump({client["Name"]: {key:
419
bool(client[key])
420
if isinstance(client[key],
421
dbus.Boolean)
422
else client[key]
423
for key in keywords}
424
for client in clients.values()},
425
fp=sys.stdout, indent=4,
426
separators=(',', ': '))
427
print()
428
else:
429
print_clients(clients.values(), keywords)
430
else:
431
# Process each client in the list by all selected options
432
for client in clients:
433
434
def set_client_prop(prop, value):
435
"""Set a Client D-Bus property"""
436
client.Set(client_interface, prop, value,
437
dbus_interface=dbus.PROPERTIES_IFACE)
438
439
def set_client_prop_ms(prop, value):
440
"""Set a Client D-Bus property, converted
441
from a string to milliseconds."""
442
set_client_prop(prop,
443
string_to_delta(value).total_seconds()
444
* 1000)
445
446
if options.remove:
447
mandos_serv.RemoveClient(client.__dbus_object_path__)
448
if options.enable:
449
set_client_prop("Enabled", dbus.Boolean(True))
450
if options.disable:
451
set_client_prop("Enabled", dbus.Boolean(False))
452
if options.bump_timeout:
453
set_client_prop("LastCheckedOK", "")
454
if options.start_checker:
455
set_client_prop("CheckerRunning", dbus.Boolean(True))
456
if options.stop_checker:
457
set_client_prop("CheckerRunning", dbus.Boolean(False))
458
if options.is_enabled:
459
if client.Get(client_interface, "Enabled",
460
dbus_interface=dbus.PROPERTIES_IFACE):
461
sys.exit(0)
462
else:
463
sys.exit(1)
464
if options.checker is not None:
465
set_client_prop("Checker", options.checker)
466
if options.host is not None:
467
set_client_prop("Host", options.host)
468
if options.interval is not None:
469
set_client_prop_ms("Interval", options.interval)
470
if options.approval_delay is not None:
471
set_client_prop_ms("ApprovalDelay",
472
options.approval_delay)
473
if options.approval_duration is not None:
474
set_client_prop_ms("ApprovalDuration",
475
options.approval_duration)
476
if options.timeout is not None:
477
set_client_prop_ms("Timeout", options.timeout)
478
if options.extended_timeout is not None:
479
set_client_prop_ms("ExtendedTimeout",
480
options.extended_timeout)
481
if options.secret is not None:
482
set_client_prop("Secret",
483
dbus.ByteArray(options.secret.read()))
484
if options.approved_by_default is not None:
485
set_client_prop("ApprovedByDefault",
486
dbus.Boolean(options
487
.approved_by_default))
488
if options.approve:
489
client.Approve(dbus.Boolean(True),
490
dbus_interface=client_interface)
491
elif options.deny:
492
client.Approve(dbus.Boolean(False),
493
dbus_interface=client_interface)
494
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
495
1689
496
1690
if __name__ == "__main__":
497
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