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