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