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:
- .bzrignore
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
import dbus
33
36
import argparse
34
37
import locale
35
38
import datetime
36
39
import re
37
40
import os
41
import collections
42
import json
43
import unittest
44
import logging
45
import io
46
import tempfile
47
import contextlib
48
49
import dbus
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
38
64
39
65
locale.setlocale(locale.LC_ALL, "")
40
66
41
tablewords = {
42
"Name": "Name",
43
"Enabled": "Enabled",
44
"Timeout": "Timeout",
45
"LastCheckedOK": "Last Successful Check",
46
"LastApprovalRequest": "Last Approval Request",
47
"Created": "Created",
48
"Interval": "Interval",
49
"Host": "Host",
50
"Fingerprint": "Fingerprint",
51
"CheckerRunning": "Check Is Running",
52
"LastEnabled": "Last Enabled",
53
"ApprovalPending": "Approval Is Pending",
54
"ApprovedByDefault": "Approved By Default",
55
"ApprovalDelay": "Approval Delay",
56
"ApprovalDuration": "Approval Duration",
57
"Checker": "Checker",
58
"ExtendedTimeout" : "Extended Timeout"
59
}
60
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
61
domain = "se.recompile"
62
busname = domain + ".Mandos"
63
server_path = "/"
64
server_interface = domain + ".Mandos"
65
client_interface = domain + ".Mandos.Client"
66
version = "1.5.3"
67
68
def timedelta_to_milliseconds(td):
69
"""Convert a datetime.timedelta object to milliseconds"""
70
return ((td.days * 24 * 60 * 60 * 1000)
71
+ (td.seconds * 1000)
72
+ (td.microseconds // 1000))
73
74
def milliseconds_to_string(ms):
75
td = datetime.timedelta(0, 0, 0, ms)
76
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
77
.format(days = "{0}T".format(td.days) if td.days else "",
78
hours = td.seconds // 3600,
79
minutes = (td.seconds % 3600) // 60,
80
seconds = td.seconds % 60,
81
))
82
83
def string_to_delta(interval):
84
"""Parse a string and return a datetime.timedelta
85
86
>>> string_to_delta("7d")
87
datetime.timedelta(7)
88
>>> string_to_delta("60s")
89
datetime.timedelta(0, 60)
90
>>> string_to_delta("60m")
91
datetime.timedelta(0, 3600)
92
>>> string_to_delta("24h")
93
datetime.timedelta(1)
94
>>> string_to_delta("1w")
95
datetime.timedelta(7)
96
>>> string_to_delta("5m 30s")
97
datetime.timedelta(0, 330)
98
"""
99
value = datetime.timedelta(0)
100
regexp = re.compile("(\d+)([dsmhw]?)")
101
102
for num, suffix in regexp.findall(interval):
103
if suffix == "d":
104
value += datetime.timedelta(int(num))
105
elif suffix == "s":
106
value += datetime.timedelta(0, int(num))
107
elif suffix == "m":
108
value += datetime.timedelta(0, 0, 0, 0, int(num))
109
elif suffix == "h":
110
value += datetime.timedelta(0, 0, 0, 0, 0, int(num))
111
elif suffix == "w":
112
value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))
113
elif suffix == "":
114
value += datetime.timedelta(0, 0, 0, int(num))
115
return value
116
117
def print_clients(clients, keywords):
118
def valuetostring(value, keyword):
119
if type(value) is dbus.Boolean:
120
return "Yes" if value else "No"
121
if keyword in ("Timeout", "Interval", "ApprovalDelay",
122
"ApprovalDuration", "ExtendedTimeout"):
123
return milliseconds_to_string(value)
124
return unicode(value)
125
126
# Create format string to print table rows
127
format_string = " ".join("{{{key}:{width}}}".format(
128
width = max(len(tablewords[key]),
129
max(len(valuetostring(client[key],
130
key))
131
for client in
132
clients)),
133
key = key) for key in keywords)
134
# Print header line
135
print(format_string.format(**tablewords))
136
for client in clients:
137
print(format_string.format(**dict((key,
138
valuetostring(client[key],
139
key))
140
for key in keywords)))
141
142
def has_actions(options):
143
return any((options.enable,
144
options.disable,
145
options.bump_timeout,
146
options.start_checker,
147
options.stop_checker,
148
options.is_enabled,
149
options.remove,
150
options.checker is not None,
151
options.timeout is not None,
152
options.extended_timeout is not None,
153
options.interval is not None,
154
options.approved_by_default is not None,
155
options.approval_delay is not None,
156
options.approval_duration is not None,
157
options.host is not None,
158
options.secret is not None,
159
options.approve,
160
options.deny))
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
161
81
162
82
def main():
163
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):
164
144
parser.add_argument("--version", action="version",
165
version = "%(prog)s {0}".format(version),
145
version="%(prog)s {}".format(version),
166
146
help="show version number and exit")
167
147
parser.add_argument("-a", "--all", action="store_true",
168
148
help="Select all clients")
169
149
parser.add_argument("-v", "--verbose", action="store_true",
170
150
help="Print all fields")
171
parser.add_argument("-e", "--enable", action="store_true",
172
help="Enable client")
173
parser.add_argument("-d", "--disable", action="store_true",
174
help="disable client")
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")
175
159
parser.add_argument("-b", "--bump-timeout", action="store_true",
176
160
help="Bump timeout for client")
177
parser.add_argument("--start-checker", action="store_true",
178
help="Start checker for client")
179
parser.add_argument("--stop-checker", action="store_true",
180
help="Stop checker 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")
181
168
parser.add_argument("-V", "--is-enabled", action="store_true",
182
169
help="Check if client is enabled")
183
170
parser.add_argument("-r", "--remove", action="store_true",
184
171
help="Remove client")
185
172
parser.add_argument("-c", "--checker",
186
173
help="Set checker command for client")
187
parser.add_argument("-t", "--timeout",
174
parser.add_argument("-t", "--timeout", type=string_to_delta,
188
175
help="Set timeout for client")
189
parser.add_argument("--extended-timeout",
176
parser.add_argument("--extended-timeout", type=string_to_delta,
190
177
help="Set extended timeout for client")
191
parser.add_argument("-i", "--interval",
178
parser.add_argument("-i", "--interval", type=string_to_delta,
192
179
help="Set checker interval for client")
193
parser.add_argument("--approve-by-default", action="store_true",
194
default=None, dest="approved_by_default",
195
help="Set client to be approved by default")
196
parser.add_argument("--deny-by-default", action="store_false",
197
dest="approved_by_default",
198
help="Set client to be denied by default")
199
parser.add_argument("--approval-delay",
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,
200
190
help="Set delay before client approve/deny")
201
parser.add_argument("--approval-duration",
191
parser.add_argument("--approval-duration", type=string_to_delta,
202
192
help="Set duration of one client approval")
203
193
parser.add_argument("-H", "--host", help="Set host for client")
204
parser.add_argument("-s", "--secret", type=file,
194
parser.add_argument("-s", "--secret",
195
type=argparse.FileType(mode="rb"),
205
196
help="Set password blob (file) for client")
206
parser.add_argument("-A", "--approve", action="store_true",
207
help="Approve any current client request")
208
parser.add_argument("-D", "--deny", action="store_true",
209
help="Deny any current client request")
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")
210
207
parser.add_argument("client", nargs="*", help="Client name")
211
options = parser.parse_args()
212
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)
219
220
221
def rfc3339_duration_to_delta(duration):
222
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
223
224
>>> rfc3339_duration_to_delta("P7D")
225
datetime.timedelta(7)
226
>>> rfc3339_duration_to_delta("PT60S")
227
datetime.timedelta(0, 60)
228
>>> rfc3339_duration_to_delta("PT60M")
229
datetime.timedelta(0, 3600)
230
>>> rfc3339_duration_to_delta("P60M")
231
datetime.timedelta(1680)
232
>>> rfc3339_duration_to_delta("PT24H")
233
datetime.timedelta(1)
234
>>> rfc3339_duration_to_delta("P1W")
235
datetime.timedelta(7)
236
>>> rfc3339_duration_to_delta("PT5M30S")
237
datetime.timedelta(0, 330)
238
>>> rfc3339_duration_to_delta("P1DT3M20S")
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'
269
"""
270
271
# Parsing an RFC 3339 duration with regular expressions is not
272
# possible - there would have to be multiple places for the same
273
# values, like seconds. The current code, while more esoteric, is
274
# cleaner without depending on a parsing library. If Python had a
275
# built-in library for parsing we would use it, but we'd like to
276
# avoid excessive use of external libraries.
277
278
# New type for defining tokens, syntax, and semantics all-in-one
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
284
# RFC 3339 "duration" tokens, syntax, and semantics; taken from
285
# the "duration" ABNF definition in RFC 3339, Appendix A.
286
token_end = Token(re.compile(r"$"), None, frozenset())
287
token_second = Token(re.compile(r"(\d+)S"),
288
datetime.timedelta(seconds=1),
289
frozenset((token_end, )))
290
token_minute = Token(re.compile(r"(\d+)M"),
291
datetime.timedelta(minutes=1),
292
frozenset((token_second, token_end)))
293
token_hour = Token(re.compile(r"(\d+)H"),
294
datetime.timedelta(hours=1),
295
frozenset((token_minute, token_end)))
296
token_time = Token(re.compile(r"T"),
297
None,
298
frozenset((token_hour, token_minute,
299
token_second)))
300
token_day = Token(re.compile(r"(\d+)D"),
301
datetime.timedelta(days=1),
302
frozenset((token_time, token_end)))
303
token_month = Token(re.compile(r"(\d+)M"),
304
datetime.timedelta(weeks=4),
305
frozenset((token_day, token_end)))
306
token_year = Token(re.compile(r"(\d+)Y"),
307
datetime.timedelta(weeks=52),
308
frozenset((token_month, token_end)))
309
token_week = Token(re.compile(r"(\d+)W"),
310
datetime.timedelta(weeks=1),
311
frozenset((token_end, )))
312
token_duration = Token(re.compile(r"P"), None,
313
frozenset((token_year, token_month,
314
token_day, token_time,
315
token_week)))
316
# Define starting values:
317
# Value so far
318
value = datetime.timedelta()
319
found_token = None
320
# Following valid tokens
321
followers = frozenset((token_duration, ))
322
# String left to parse
323
s = duration
324
# Loop until end token is found
325
while found_token is not token_end:
326
# Search for any currently valid tokens
327
for token in followers:
328
match = token.regexp.match(s)
329
if match is not None:
330
# Token found
331
if token.value is not None:
332
# Value found, parse digits
333
factor = int(match.group(1), 10)
334
# Add to value so far
335
value += factor * token.value
336
# Strip token from string
337
s = token.regexp.sub("", s, 1)
338
# Go to found token
339
found_token = token
340
# Set valid next tokens
341
followers = found_token.followers
342
break
343
else:
344
# No currently valid tokens were found
345
raise ValueError("Invalid RFC 3339 duration: {!r}"
346
.format(duration))
347
# End token found
348
return value
349
350
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')
356
datetime.timedelta(7)
357
>>> parse_pre_1_6_1_interval('60s')
358
datetime.timedelta(0, 60)
359
>>> parse_pre_1_6_1_interval('60m')
360
datetime.timedelta(0, 3600)
361
>>> parse_pre_1_6_1_interval('24h')
362
datetime.timedelta(1)
363
>>> parse_pre_1_6_1_interval('1w')
364
datetime.timedelta(7)
365
>>> parse_pre_1_6_1_interval('5m 30s')
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
373
"""
374
375
value = datetime.timedelta(0)
376
regexp = re.compile(r"(\d+)([dsmhw]?)")
377
378
for num, suffix in regexp.findall(interval):
379
if suffix == "d":
380
value += datetime.timedelta(int(num))
381
elif suffix == "s":
382
value += datetime.timedelta(0, int(num))
383
elif suffix == "m":
384
value += datetime.timedelta(0, 0, 0, 0, int(num))
385
elif suffix == "h":
386
value += datetime.timedelta(0, 0, 0, 0, 0, int(num))
387
elif suffix == "w":
388
value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))
389
elif suffix == "":
390
value += datetime.timedelta(0, 0, 0, int(num))
391
return value
392
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
213
418
if has_actions(options) and not (options.client or options.all):
214
419
parser.error("Options require clients names or --all.")
215
420
if options.verbose and has_actions(options):
216
parser.error("--verbose can only be used alone or with"
217
" --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.")
218
425
if options.all and not has_actions(options):
219
426
parser.error("--all requires an action.")
220
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):
221
437
try:
222
bus = dbus.SystemBus()
223
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)
224
442
except dbus.exceptions.DBusException:
225
print("Could not connect to Mandos server",
226
file=sys.stderr)
227
sys.exit(1)
228
229
mandos_serv = dbus.Interface(mandos_dbus_objc,
230
dbus_interface = server_interface)
231
232
#block stderr since dbus library prints to stderr
233
null = os.open(os.path.devnull, os.O_RDWR)
234
stderrcopy = os.dup(sys.stderr.fileno())
235
os.dup2(null, sys.stderr.fileno())
236
os.close(null)
237
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)
238
924
try:
239
mandos_clients = mandos_serv.GetAllClientsWithProperties()
925
yield
240
926
finally:
241
#restore stderr
927
# restore stderr
242
928
os.dup2(stderrcopy, sys.stderr.fileno())
243
929
os.close(stderrcopy)
244
except dbus.exceptions.DBusException:
245
print("Access denied: Accessing mandos server through dbus.",
246
file=sys.stderr)
247
sys.exit(1)
248
249
# Compile dict of (clients: properties) to process
250
clients={}
251
252
if options.all or not options.client:
253
clients = dict((bus.get_object(busname, path), properties)
254
for path, properties in
255
mandos_clients.iteritems())
256
else:
257
for name in options.client:
258
for path, client in mandos_clients.iteritems():
259
if client["Name"] == name:
260
client_objc = bus.get_object(busname, path)
261
clients[client_objc] = client
262
break
263
else:
264
print("Client not found on server: {0!r}"
265
.format(name), file=sys.stderr)
266
sys.exit(1)
267
268
if not has_actions(options) and clients:
269
if options.verbose:
270
keywords = ("Name", "Enabled", "Timeout",
271
"LastCheckedOK", "Created", "Interval",
272
"Host", "Fingerprint", "CheckerRunning",
273
"LastEnabled", "ApprovalPending",
274
"ApprovedByDefault",
275
"LastApprovalRequest", "ApprovalDelay",
276
"ApprovalDuration", "Checker",
277
"ExtendedTimeout")
278
else:
279
keywords = defaultkeywords
280
281
print_clients(clients.values(), keywords)
282
else:
283
# Process each client in the list by all selected options
284
for client in clients:
285
def set_client_prop(prop, value):
286
"""Set a Client D-Bus property"""
287
client.Set(client_interface, prop, value,
288
dbus_interface=dbus.PROPERTIES_IFACE)
289
def set_client_prop_ms(prop, value):
290
"""Set a Client D-Bus property, converted
291
from a string to milliseconds."""
292
set_client_prop(prop,
293
timedelta_to_milliseconds
294
(string_to_delta(value)))
295
if options.remove:
296
mandos_serv.RemoveClient(client.__dbus_object_path__)
297
if options.enable:
298
set_client_prop("Enabled", dbus.Boolean(True))
299
if options.disable:
300
set_client_prop("Enabled", dbus.Boolean(False))
301
if options.bump_timeout:
302
set_client_prop("LastCheckedOK", "")
303
if options.start_checker:
304
set_client_prop("CheckerRunning", dbus.Boolean(True))
305
if options.stop_checker:
306
set_client_prop("CheckerRunning", dbus.Boolean(False))
307
if options.is_enabled:
308
sys.exit(0 if client.Get(client_interface,
309
"Enabled",
310
dbus_interface=
311
dbus.PROPERTIES_IFACE)
312
else 1)
313
if options.checker is not None:
314
set_client_prop("Checker", options.checker)
315
if options.host is not None:
316
set_client_prop("Host", options.host)
317
if options.interval is not None:
318
set_client_prop_ms("Interval", options.interval)
319
if options.approval_delay is not None:
320
set_client_prop_ms("ApprovalDelay",
321
options.approval_delay)
322
if options.approval_duration is not None:
323
set_client_prop_ms("ApprovalDuration",
324
options.approval_duration)
325
if options.timeout is not None:
326
set_client_prop_ms("Timeout", options.timeout)
327
if options.extended_timeout is not None:
328
set_client_prop_ms("ExtendedTimeout",
329
options.extended_timeout)
330
if options.secret is not None:
331
set_client_prop("Secret",
332
dbus.ByteArray(options.secret.read()))
333
if options.approved_by_default is not None:
334
set_client_prop("ApprovedByDefault",
335
dbus.Boolean(options
336
.approved_by_default))
337
if options.approve:
338
client.Approve(dbus.Boolean(True),
339
dbus_interface=client_interface)
340
elif options.deny:
341
client.Approve(dbus.Boolean(False),
342
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
343
1787
344
1788
if __name__ == "__main__":
345
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