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