/mandos/release

To get this branch, use:
bzr branch http://bzr.recompile.se/loggerhead/mandos/release

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-31 04:47:26 UTC
  • mto: This revision was merged to the branch mainline in revision 382.
  • Revision ID: teddy@recompile.se-20190331044726-po11kmjmug07mw00
mandos-ctl: Remove import of unused module

* mandos-ctl: Remove "import abc".

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
#!/usr/bin/python
 
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
# Mandos Monitor - Control and monitor the Mandos server
 
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
 
13
# the Free Software Foundation, either version 3 of the License, or
 
14
# (at your option) any later version.
 
15
#
 
16
#     Mandos is distributed in the hope that it will be useful, but
 
17
#     WITHOUT ANY WARRANTY; without even the implied warranty of
 
18
#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
19
#     GNU General Public License for more details.
 
20
#
 
21
# You should have received a copy of the GNU General Public License
 
22
# along with Mandos.  If not, see <http://www.gnu.org/licenses/>.
 
23
#
 
24
# Contact the authors at <mandos@recompile.se>.
 
25
#
 
26
 
 
27
from __future__ import (division, absolute_import, print_function,
 
28
                        unicode_literals)
 
29
 
 
30
try:
 
31
    from future_builtins import *
 
32
except ImportError:
 
33
    pass
 
34
 
 
35
import sys
 
36
import argparse
 
37
import locale
 
38
import datetime
 
39
import re
 
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 as dbus_python
 
50
 
 
51
# Show warnings by default
 
52
if not sys.warnoptions:
 
53
    import warnings
 
54
    warnings.simplefilter("default")
 
55
 
 
56
log = logging.getLogger(sys.argv[0])
 
57
logging.basicConfig(level="INFO", # Show info level messages
 
58
                    format="%(message)s") # Show basic log messages
 
59
 
 
60
logging.captureWarnings(True)   # Show warnings via the logging system
 
61
 
 
62
if sys.version_info.major == 2:
 
63
    str = unicode
 
64
    import StringIO
 
65
    io.StringIO = StringIO.StringIO
 
66
 
 
67
locale.setlocale(locale.LC_ALL, "")
 
68
 
 
69
version = "1.8.3"
 
70
 
 
71
 
 
72
def main():
 
73
    parser = argparse.ArgumentParser()
 
74
    add_command_line_options(parser)
 
75
 
 
76
    options = parser.parse_args()
 
77
    check_option_syntax(parser, options)
 
78
 
 
79
    clientnames = options.client
 
80
 
 
81
    if options.debug:
 
82
        log.setLevel(logging.DEBUG)
 
83
 
 
84
    bus = dbus_python_adapter.CachingBus(dbus_python)
 
85
 
 
86
    try:
 
87
        all_clients = bus.get_clients_and_properties()
 
88
    except dbus.ConnectFailed as e:
 
89
        log.critical("Could not connect to Mandos server: %s", e)
 
90
        sys.exit(1)
 
91
    except dbus.Error as e:
 
92
        log.critical(
 
93
            "Failed to access Mandos server through D-Bus:\n%s", e)
 
94
        sys.exit(1)
 
95
 
 
96
    # Compile dict of (clientpath: properties) to process
 
97
    if not clientnames:
 
98
        clients = all_clients
 
99
    else:
 
100
        clients = {}
 
101
        for name in clientnames:
 
102
            for objpath, properties in all_clients.items():
 
103
                if properties["Name"] == name:
 
104
                    clients[objpath] = properties
 
105
                    break
 
106
            else:
 
107
                log.critical("Client not found on server: %r", name)
 
108
                sys.exit(1)
 
109
 
 
110
    commands = commands_from_options(options)
 
111
 
 
112
    for command in commands:
 
113
        command.run(clients, bus)
 
114
 
 
115
 
 
116
def add_command_line_options(parser):
 
117
    parser.add_argument("--version", action="version",
 
118
                        version="%(prog)s {}".format(version),
 
119
                        help="show version number and exit")
 
120
    parser.add_argument("-a", "--all", action="store_true",
 
121
                        help="Select all clients")
 
122
    parser.add_argument("-v", "--verbose", action="store_true",
 
123
                        help="Print all fields")
 
124
    parser.add_argument("-j", "--dump-json", dest="commands",
 
125
                        action="append_const", default=[],
 
126
                        const=command.DumpJSON(),
 
127
                        help="Dump client data in JSON format")
 
128
    enable_disable = parser.add_mutually_exclusive_group()
 
129
    enable_disable.add_argument("-e", "--enable", dest="commands",
 
130
                                action="append_const", default=[],
 
131
                                const=command.Enable(),
 
132
                                help="Enable client")
 
133
    enable_disable.add_argument("-d", "--disable", dest="commands",
 
134
                                action="append_const", default=[],
 
135
                                const=command.Disable(),
 
136
                                help="disable client")
 
137
    parser.add_argument("-b", "--bump-timeout", dest="commands",
 
138
                        action="append_const", default=[],
 
139
                        const=command.BumpTimeout(),
 
140
                        help="Bump timeout for client")
 
141
    start_stop_checker = parser.add_mutually_exclusive_group()
 
142
    start_stop_checker.add_argument("--start-checker",
 
143
                                    dest="commands",
 
144
                                    action="append_const", default=[],
 
145
                                    const=command.StartChecker(),
 
146
                                    help="Start checker for client")
 
147
    start_stop_checker.add_argument("--stop-checker", dest="commands",
 
148
                                    action="append_const", default=[],
 
149
                                    const=command.StopChecker(),
 
150
                                    help="Stop checker for client")
 
151
    parser.add_argument("-V", "--is-enabled", dest="commands",
 
152
                        action="append_const", default=[],
 
153
                        const=command.IsEnabled(),
 
154
                        help="Check if client is enabled")
 
155
    parser.add_argument("-r", "--remove", dest="commands",
 
156
                        action="append_const", default=[],
 
157
                        const=command.Remove(),
 
158
                        help="Remove client")
 
159
    parser.add_argument("-c", "--checker", dest="commands",
 
160
                        action="append", default=[],
 
161
                        metavar="COMMAND", type=command.SetChecker,
 
162
                        help="Set checker command for client")
 
163
    parser.add_argument(
 
164
        "-t", "--timeout", dest="commands", action="append",
 
165
        default=[], metavar="TIME",
 
166
        type=command.SetTimeout.argparse(string_to_delta),
 
167
        help="Set timeout for client")
 
168
    parser.add_argument(
 
169
        "--extended-timeout", dest="commands", action="append",
 
170
        default=[], metavar="TIME",
 
171
        type=command.SetExtendedTimeout.argparse(string_to_delta),
 
172
        help="Set extended timeout for client")
 
173
    parser.add_argument(
 
174
        "-i", "--interval", dest="commands", action="append",
 
175
        default=[], metavar="TIME",
 
176
        type=command.SetInterval.argparse(string_to_delta),
 
177
        help="Set checker interval for client")
 
178
    approve_deny_default = parser.add_mutually_exclusive_group()
 
179
    approve_deny_default.add_argument(
 
180
        "--approve-by-default", dest="commands",
 
181
        action="append_const", default=[],
 
182
        const=command.ApproveByDefault(),
 
183
        help="Set client to be approved by default")
 
184
    approve_deny_default.add_argument(
 
185
        "--deny-by-default", dest="commands",
 
186
        action="append_const", default=[],
 
187
        const=command.DenyByDefault(),
 
188
        help="Set client to be denied by default")
 
189
    parser.add_argument(
 
190
        "--approval-delay", dest="commands", action="append",
 
191
        default=[], metavar="TIME",
 
192
        type=command.SetApprovalDelay.argparse(string_to_delta),
 
193
        help="Set delay before client approve/deny")
 
194
    parser.add_argument(
 
195
        "--approval-duration", dest="commands", action="append",
 
196
        default=[], metavar="TIME",
 
197
        type=command.SetApprovalDuration.argparse(string_to_delta),
 
198
        help="Set duration of one client approval")
 
199
    parser.add_argument("-H", "--host", dest="commands",
 
200
                        action="append", default=[], metavar="STRING",
 
201
                        type=command.SetHost,
 
202
                        help="Set host for client")
 
203
    parser.add_argument(
 
204
        "-s", "--secret", dest="commands", action="append",
 
205
        default=[], metavar="FILENAME",
 
206
        type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
 
207
        help="Set password blob (file) for client")
 
208
    approve_deny = parser.add_mutually_exclusive_group()
 
209
    approve_deny.add_argument(
 
210
        "-A", "--approve", dest="commands", action="append_const",
 
211
        default=[], const=command.Approve(),
 
212
        help="Approve any current client request")
 
213
    approve_deny.add_argument("-D", "--deny", dest="commands",
 
214
                              action="append_const", default=[],
 
215
                              const=command.Deny(),
 
216
                              help="Deny any current client request")
 
217
    parser.add_argument("--debug", action="store_true",
 
218
                        help="Debug mode (show D-Bus commands)")
 
219
    parser.add_argument("--check", action="store_true",
 
220
                        help="Run self-test")
 
221
    parser.add_argument("client", nargs="*", help="Client name")
 
222
 
 
223
 
 
224
def string_to_delta(interval):
 
225
    """Parse a string and return a datetime.timedelta"""
 
226
 
 
227
    try:
 
228
        return rfc3339_duration_to_delta(interval)
 
229
    except ValueError as e:
 
230
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
 
231
                    ' '.join(e.args))
 
232
    return parse_pre_1_6_1_interval(interval)
 
233
 
 
234
 
 
235
def rfc3339_duration_to_delta(duration):
 
236
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
 
237
 
 
238
    >>> rfc3339_duration_to_delta("P7D")
 
239
    datetime.timedelta(7)
 
240
    >>> rfc3339_duration_to_delta("PT60S")
 
241
    datetime.timedelta(0, 60)
 
242
    >>> rfc3339_duration_to_delta("PT60M")
 
243
    datetime.timedelta(0, 3600)
 
244
    >>> rfc3339_duration_to_delta("P60M")
 
245
    datetime.timedelta(1680)
 
246
    >>> rfc3339_duration_to_delta("PT24H")
 
247
    datetime.timedelta(1)
 
248
    >>> rfc3339_duration_to_delta("P1W")
 
249
    datetime.timedelta(7)
 
250
    >>> rfc3339_duration_to_delta("PT5M30S")
 
251
    datetime.timedelta(0, 330)
 
252
    >>> rfc3339_duration_to_delta("P1DT3M20S")
 
253
    datetime.timedelta(1, 200)
 
254
    >>> # Can not be empty:
 
255
    >>> rfc3339_duration_to_delta("")
 
256
    Traceback (most recent call last):
 
257
    ...
 
258
    ValueError: Invalid RFC 3339 duration: ""
 
259
    >>> # Must start with "P":
 
260
    >>> rfc3339_duration_to_delta("1D")
 
261
    Traceback (most recent call last):
 
262
    ...
 
263
    ValueError: Invalid RFC 3339 duration: "1D"
 
264
    >>> # Must use correct order
 
265
    >>> rfc3339_duration_to_delta("PT1S2M")
 
266
    Traceback (most recent call last):
 
267
    ...
 
268
    ValueError: Invalid RFC 3339 duration: "PT1S2M"
 
269
    >>> # Time needs time marker
 
270
    >>> rfc3339_duration_to_delta("P1H2S")
 
271
    Traceback (most recent call last):
 
272
    ...
 
273
    ValueError: Invalid RFC 3339 duration: "P1H2S"
 
274
    >>> # Weeks can not be combined with anything else
 
275
    >>> rfc3339_duration_to_delta("P1D2W")
 
276
    Traceback (most recent call last):
 
277
    ...
 
278
    ValueError: Invalid RFC 3339 duration: "P1D2W"
 
279
    >>> rfc3339_duration_to_delta("P2W2H")
 
280
    Traceback (most recent call last):
 
281
    ...
 
282
    ValueError: Invalid RFC 3339 duration: "P2W2H"
 
283
    """
 
284
 
 
285
    # Parsing an RFC 3339 duration with regular expressions is not
 
286
    # possible - there would have to be multiple places for the same
 
287
    # values, like seconds.  The current code, while more esoteric, is
 
288
    # cleaner without depending on a parsing library.  If Python had a
 
289
    # built-in library for parsing we would use it, but we'd like to
 
290
    # avoid excessive use of external libraries.
 
291
 
 
292
    # New type for defining tokens, syntax, and semantics all-in-one
 
293
    Token = collections.namedtuple("Token", (
 
294
        "regexp",  # To match token; if "value" is not None, must have
 
295
                   # a "group" containing digits
 
296
        "value",   # datetime.timedelta or None
 
297
        "followers"))           # Tokens valid after this token
 
298
    # RFC 3339 "duration" tokens, syntax, and semantics; taken from
 
299
    # the "duration" ABNF definition in RFC 3339, Appendix A.
 
300
    token_end = Token(re.compile(r"$"), None, frozenset())
 
301
    token_second = Token(re.compile(r"(\d+)S"),
 
302
                         datetime.timedelta(seconds=1),
 
303
                         frozenset((token_end, )))
 
304
    token_minute = Token(re.compile(r"(\d+)M"),
 
305
                         datetime.timedelta(minutes=1),
 
306
                         frozenset((token_second, token_end)))
 
307
    token_hour = Token(re.compile(r"(\d+)H"),
 
308
                       datetime.timedelta(hours=1),
 
309
                       frozenset((token_minute, token_end)))
 
310
    token_time = Token(re.compile(r"T"),
 
311
                       None,
 
312
                       frozenset((token_hour, token_minute,
 
313
                                  token_second)))
 
314
    token_day = Token(re.compile(r"(\d+)D"),
 
315
                      datetime.timedelta(days=1),
 
316
                      frozenset((token_time, token_end)))
 
317
    token_month = Token(re.compile(r"(\d+)M"),
 
318
                        datetime.timedelta(weeks=4),
 
319
                        frozenset((token_day, token_end)))
 
320
    token_year = Token(re.compile(r"(\d+)Y"),
 
321
                       datetime.timedelta(weeks=52),
 
322
                       frozenset((token_month, token_end)))
 
323
    token_week = Token(re.compile(r"(\d+)W"),
 
324
                       datetime.timedelta(weeks=1),
 
325
                       frozenset((token_end, )))
 
326
    token_duration = Token(re.compile(r"P"), None,
 
327
                           frozenset((token_year, token_month,
 
328
                                      token_day, token_time,
 
329
                                      token_week)))
 
330
    # Define starting values:
 
331
    # Value so far
 
332
    value = datetime.timedelta()
 
333
    found_token = None
 
334
    # Following valid tokens
 
335
    followers = frozenset((token_duration, ))
 
336
    # String left to parse
 
337
    s = duration
 
338
    # Loop until end token is found
 
339
    while found_token is not token_end:
 
340
        # Search for any currently valid tokens
 
341
        for token in followers:
 
342
            match = token.regexp.match(s)
 
343
            if match is not None:
 
344
                # Token found
 
345
                if token.value is not None:
 
346
                    # Value found, parse digits
 
347
                    factor = int(match.group(1), 10)
 
348
                    # Add to value so far
 
349
                    value += factor * token.value
 
350
                # Strip token from string
 
351
                s = token.regexp.sub("", s, 1)
 
352
                # Go to found token
 
353
                found_token = token
 
354
                # Set valid next tokens
 
355
                followers = found_token.followers
 
356
                break
 
357
        else:
 
358
            # No currently valid tokens were found
 
359
            raise ValueError("Invalid RFC 3339 duration: \"{}\""
 
360
                             .format(duration))
 
361
    # End token found
 
362
    return value
 
363
 
 
364
 
 
365
def parse_pre_1_6_1_interval(interval):
 
366
    """Parse an interval string as documented by Mandos before 1.6.1,
 
367
    and return a datetime.timedelta
 
368
 
 
369
    >>> parse_pre_1_6_1_interval('7d')
 
370
    datetime.timedelta(7)
 
371
    >>> parse_pre_1_6_1_interval('60s')
 
372
    datetime.timedelta(0, 60)
 
373
    >>> parse_pre_1_6_1_interval('60m')
 
374
    datetime.timedelta(0, 3600)
 
375
    >>> parse_pre_1_6_1_interval('24h')
 
376
    datetime.timedelta(1)
 
377
    >>> parse_pre_1_6_1_interval('1w')
 
378
    datetime.timedelta(7)
 
379
    >>> parse_pre_1_6_1_interval('5m 30s')
 
380
    datetime.timedelta(0, 330)
 
381
    >>> parse_pre_1_6_1_interval('')
 
382
    datetime.timedelta(0)
 
383
    >>> # Ignore unknown characters, allow any order and repetitions
 
384
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
 
385
    datetime.timedelta(2, 480, 18000)
 
386
 
 
387
    """
 
388
 
 
389
    value = datetime.timedelta(0)
 
390
    regexp = re.compile(r"(\d+)([dsmhw]?)")
 
391
 
 
392
    for num, suffix in regexp.findall(interval):
 
393
        if suffix == "d":
 
394
            value += datetime.timedelta(int(num))
 
395
        elif suffix == "s":
 
396
            value += datetime.timedelta(0, int(num))
 
397
        elif suffix == "m":
 
398
            value += datetime.timedelta(0, 0, 0, 0, int(num))
 
399
        elif suffix == "h":
 
400
            value += datetime.timedelta(0, 0, 0, 0, 0, int(num))
 
401
        elif suffix == "w":
 
402
            value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))
 
403
        elif suffix == "":
 
404
            value += datetime.timedelta(0, 0, 0, int(num))
 
405
    return value
 
406
 
 
407
 
 
408
def check_option_syntax(parser, options):
 
409
    """Apply additional restrictions on options, not expressible in
 
410
argparse"""
 
411
 
 
412
    def has_commands(options, commands=None):
 
413
        if commands is None:
 
414
            commands = (command.Enable,
 
415
                        command.Disable,
 
416
                        command.BumpTimeout,
 
417
                        command.StartChecker,
 
418
                        command.StopChecker,
 
419
                        command.IsEnabled,
 
420
                        command.Remove,
 
421
                        command.SetChecker,
 
422
                        command.SetTimeout,
 
423
                        command.SetExtendedTimeout,
 
424
                        command.SetInterval,
 
425
                        command.ApproveByDefault,
 
426
                        command.DenyByDefault,
 
427
                        command.SetApprovalDelay,
 
428
                        command.SetApprovalDuration,
 
429
                        command.SetHost,
 
430
                        command.SetSecret,
 
431
                        command.Approve,
 
432
                        command.Deny)
 
433
        return any(isinstance(cmd, commands)
 
434
                   for cmd in options.commands)
 
435
 
 
436
    if has_commands(options) and not (options.client or options.all):
 
437
        parser.error("Options require clients names or --all.")
 
438
    if options.verbose and has_commands(options):
 
439
        parser.error("--verbose can only be used alone.")
 
440
    if (has_commands(options, (command.DumpJSON,))
 
441
        and (options.verbose or len(options.commands) > 1)):
 
442
        parser.error("--dump-json can only be used alone.")
 
443
    if options.all and not has_commands(options):
 
444
        parser.error("--all requires an action.")
 
445
    if (has_commands(options, (command.IsEnabled,))
 
446
        and len(options.client) > 1):
 
447
        parser.error("--is-enabled requires exactly one client")
 
448
    if (len(options.commands) > 1
 
449
        and has_commands(options, (command.Remove,))
 
450
        and not has_commands(options, (command.Deny,))):
 
451
        parser.error("--remove can only be combined with --deny")
 
452
 
 
453
 
 
454
class dbus(object):
 
455
 
 
456
    class SystemBus(object):
 
457
 
 
458
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
 
459
        def get_managed_objects(self, busname, objectpath):
 
460
            return self.call_method("GetManagedObjects", busname,
 
461
                                    objectpath,
 
462
                                    self.object_manager_iface)
 
463
 
 
464
        properties_iface = "org.freedesktop.DBus.Properties"
 
465
        def set_property(self, busname, objectpath, interface, key,
 
466
                         value):
 
467
            self.call_method("Set", busname, objectpath,
 
468
                             self.properties_iface, interface, key,
 
469
                             value)
 
470
 
 
471
 
 
472
    class MandosBus(SystemBus):
 
473
        busname_domain = "se.recompile"
 
474
        busname = busname_domain + ".Mandos"
 
475
        server_path = "/"
 
476
        server_interface = busname_domain + ".Mandos"
 
477
        client_interface = busname_domain + ".Mandos.Client"
 
478
        del busname_domain
 
479
 
 
480
        def get_clients_and_properties(self):
 
481
            managed_objects = self.get_managed_objects(
 
482
                self.busname, self.server_path)
 
483
            return {objpath: properties[self.client_interface]
 
484
                    for objpath, properties in managed_objects.items()
 
485
                    if self.client_interface in properties}
 
486
 
 
487
        def set_client_property(self, objectpath, key, value):
 
488
            return self.set_property(self.busname, objectpath,
 
489
                                     self.client_interface, key,
 
490
                                     value)
 
491
 
 
492
        def call_client_method(self, objectpath, method, *args):
 
493
            return self.call_method(method, self.busname, objectpath,
 
494
                                    self.client_interface, *args)
 
495
 
 
496
        def call_server_method(self, method, *args):
 
497
            return self.call_method(method, self.busname,
 
498
                                    self.server_path,
 
499
                                    self.server_interface, *args)
 
500
 
 
501
    class Error(Exception):
 
502
        pass
 
503
 
 
504
    class ConnectFailed(Error):
 
505
        pass
 
506
 
 
507
 
 
508
class dbus_python_adapter(object):
 
509
 
 
510
    class SystemBus(dbus.MandosBus):
 
511
        """Use dbus-python"""
 
512
 
 
513
        def __init__(self, module=dbus_python):
 
514
            self.dbus_python = module
 
515
            self.bus = self.dbus_python.SystemBus()
 
516
 
 
517
        @contextlib.contextmanager
 
518
        def convert_exception(self, exception_class=dbus.Error):
 
519
            try:
 
520
                yield
 
521
            except self.dbus_python.exceptions.DBusException as e:
 
522
                # This does what "raise from" would do
 
523
                exc = exception_class(*e.args)
 
524
                exc.__cause__ = e
 
525
                raise exc
 
526
 
 
527
        def call_method(self, methodname, busname, objectpath,
 
528
                        interface, *args):
 
529
            proxy_object = self.get_object(busname, objectpath)
 
530
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
531
                      interface, methodname,
 
532
                      ", ".join(repr(a) for a in args))
 
533
            method = getattr(proxy_object, methodname)
 
534
            with self.convert_exception():
 
535
                with dbus_python_adapter.SilenceLogger(
 
536
                        "dbus.proxies"):
 
537
                    value = method(*args, dbus_interface=interface)
 
538
            return self.type_filter(value)
 
539
 
 
540
        def get_object(self, busname, objectpath):
 
541
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
542
                      busname, objectpath)
 
543
            with self.convert_exception(dbus.ConnectFailed):
 
544
                return self.bus.get_object(busname, objectpath)
 
545
 
 
546
        def type_filter(self, value):
 
547
            """Convert the most bothersome types to Python types"""
 
548
            if isinstance(value, self.dbus_python.Boolean):
 
549
                return bool(value)
 
550
            if isinstance(value, self.dbus_python.ObjectPath):
 
551
                return str(value)
 
552
            # Also recurse into dictionaries
 
553
            if isinstance(value, self.dbus_python.Dictionary):
 
554
                return {self.type_filter(key):
 
555
                        self.type_filter(subval)
 
556
                        for key, subval in value.items()}
 
557
            return value
 
558
 
 
559
 
 
560
    class SilenceLogger(object):
 
561
        "Simple context manager to silence a particular logger"
 
562
        def __init__(self, loggername):
 
563
            self.logger = logging.getLogger(loggername)
 
564
 
 
565
        def __enter__(self):
 
566
            self.logger.addFilter(self.nullfilter)
 
567
 
 
568
        class NullFilter(logging.Filter):
 
569
            def filter(self, record):
 
570
                return False
 
571
 
 
572
        nullfilter = NullFilter()
 
573
 
 
574
        def __exit__(self, exc_type, exc_val, exc_tb):
 
575
            self.logger.removeFilter(self.nullfilter)
 
576
 
 
577
 
 
578
    class CachingBus(SystemBus):
 
579
        """A caching layer for dbus_python_adapter.SystemBus"""
 
580
        def __init__(self, *args, **kwargs):
 
581
            self.object_cache = {}
 
582
            super(dbus_python_adapter.CachingBus,
 
583
                  self).__init__(*args, **kwargs)
 
584
        def get_object(self, busname, objectpath):
 
585
            try:
 
586
                return self.object_cache[(busname, objectpath)]
 
587
            except KeyError:
 
588
                new_object = super(
 
589
                    dbus_python_adapter.CachingBus,
 
590
                    self).get_object(busname, objectpath)
 
591
                self.object_cache[(busname, objectpath)]  = new_object
 
592
                return new_object
 
593
 
 
594
 
 
595
def commands_from_options(options):
 
596
 
 
597
    commands = list(options.commands)
 
598
 
 
599
    def find_cmd(cmd, commands):
 
600
        i = 0
 
601
        for i, c in enumerate(commands):
 
602
            if isinstance(c, cmd):
 
603
                return i
 
604
        return i+1
 
605
 
 
606
    # If command.Remove is present, move any instances of command.Deny
 
607
    # to occur ahead of command.Remove.
 
608
    index_of_remove = find_cmd(command.Remove, commands)
 
609
    before_remove = commands[:index_of_remove]
 
610
    after_remove = commands[index_of_remove:]
 
611
    cleaned_after = []
 
612
    for cmd in after_remove:
 
613
        if isinstance(cmd, command.Deny):
 
614
            before_remove.append(cmd)
 
615
        else:
 
616
            cleaned_after.append(cmd)
 
617
    if cleaned_after != after_remove:
 
618
        commands = before_remove + cleaned_after
 
619
 
 
620
    # If no command option has been given, show table of clients,
 
621
    # optionally verbosely
 
622
    if not commands:
 
623
        commands.append(command.PrintTable(verbose=options.verbose))
 
624
 
 
625
    return commands
 
626
 
 
627
 
 
628
class command(object):
 
629
    """A namespace for command classes"""
 
630
 
 
631
    class Base(object):
 
632
        """Abstract base class for commands"""
 
633
        def run(self, clients, bus=None):
 
634
            """Normal commands should implement run_on_one_client(),
 
635
but commands which want to operate on all clients at the same time can
 
636
override this run() method instead.
 
637
"""
 
638
            self.bus = bus
 
639
            for client, properties in clients.items():
 
640
                self.run_on_one_client(client, properties)
 
641
 
 
642
 
 
643
    class IsEnabled(Base):
 
644
        def run(self, clients, bus=None):
 
645
            properties = next(iter(clients.values()))
 
646
            if properties["Enabled"]:
 
647
                sys.exit(0)
 
648
            sys.exit(1)
 
649
 
 
650
 
 
651
    class Approve(Base):
 
652
        def run_on_one_client(self, client, properties):
 
653
            self.bus.call_client_method(client, "Approve", True)
 
654
 
 
655
 
 
656
    class Deny(Base):
 
657
        def run_on_one_client(self, client, properties):
 
658
            self.bus.call_client_method(client, "Approve", False)
 
659
 
 
660
 
 
661
    class Remove(Base):
 
662
        def run(self, clients, bus):
 
663
            for clientpath in frozenset(clients.keys()):
 
664
                bus.call_server_method("RemoveClient", clientpath)
 
665
 
 
666
 
 
667
    class Output(Base):
 
668
        """Abstract class for commands outputting client details"""
 
669
        all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
 
670
                        "Created", "Interval", "Host", "KeyID",
 
671
                        "Fingerprint", "CheckerRunning",
 
672
                        "LastEnabled", "ApprovalPending",
 
673
                        "ApprovedByDefault", "LastApprovalRequest",
 
674
                        "ApprovalDelay", "ApprovalDuration",
 
675
                        "Checker", "ExtendedTimeout", "Expires",
 
676
                        "LastCheckerStatus")
 
677
 
 
678
 
 
679
    class DumpJSON(Output):
 
680
        def run(self, clients, bus=None):
 
681
            data = {properties["Name"]:
 
682
                    {key: properties[key]
 
683
                     for key in self.all_keywords}
 
684
                    for properties in clients.values()}
 
685
            print(json.dumps(data, indent=4, separators=(',', ': ')))
 
686
 
 
687
 
 
688
    class PrintTable(Output):
 
689
        def __init__(self, verbose=False):
 
690
            self.verbose = verbose
 
691
 
 
692
        def run(self, clients, bus=None):
 
693
            default_keywords = ("Name", "Enabled", "Timeout",
 
694
                                "LastCheckedOK")
 
695
            keywords = default_keywords
 
696
            if self.verbose:
 
697
                keywords = self.all_keywords
 
698
            print(self.TableOfClients(clients.values(), keywords))
 
699
 
 
700
        class TableOfClients(object):
 
701
            tableheaders = {
 
702
                "Name": "Name",
 
703
                "Enabled": "Enabled",
 
704
                "Timeout": "Timeout",
 
705
                "LastCheckedOK": "Last Successful Check",
 
706
                "LastApprovalRequest": "Last Approval Request",
 
707
                "Created": "Created",
 
708
                "Interval": "Interval",
 
709
                "Host": "Host",
 
710
                "Fingerprint": "Fingerprint",
 
711
                "KeyID": "Key ID",
 
712
                "CheckerRunning": "Check Is Running",
 
713
                "LastEnabled": "Last Enabled",
 
714
                "ApprovalPending": "Approval Is Pending",
 
715
                "ApprovedByDefault": "Approved By Default",
 
716
                "ApprovalDelay": "Approval Delay",
 
717
                "ApprovalDuration": "Approval Duration",
 
718
                "Checker": "Checker",
 
719
                "ExtendedTimeout": "Extended Timeout",
 
720
                "Expires": "Expires",
 
721
                "LastCheckerStatus": "Last Checker Status",
 
722
            }
 
723
 
 
724
            def __init__(self, clients, keywords):
 
725
                self.clients = clients
 
726
                self.keywords = keywords
 
727
 
 
728
            def __str__(self):
 
729
                return "\n".join(self.rows())
 
730
 
 
731
            if sys.version_info.major == 2:
 
732
                __unicode__ = __str__
 
733
                def __str__(self):
 
734
                    return str(self).encode(
 
735
                        locale.getpreferredencoding())
 
736
 
 
737
            def rows(self):
 
738
                format_string = self.row_formatting_string()
 
739
                rows = [self.header_line(format_string)]
 
740
                rows.extend(self.client_line(client, format_string)
 
741
                            for client in self.clients)
 
742
                return rows
 
743
 
 
744
            def row_formatting_string(self):
 
745
                "Format string used to format table rows"
 
746
                return " ".join("{{{key}:{width}}}".format(
 
747
                    width=max(len(self.tableheaders[key]),
 
748
                              *(len(self.string_from_client(client,
 
749
                                                            key))
 
750
                                for client in self.clients)),
 
751
                    key=key)
 
752
                                for key in self.keywords)
 
753
 
 
754
            def string_from_client(self, client, key):
 
755
                return self.valuetostring(client[key], key)
 
756
 
 
757
            @classmethod
 
758
            def valuetostring(cls, value, keyword):
 
759
                if isinstance(value, bool):
 
760
                    return "Yes" if value else "No"
 
761
                if keyword in ("Timeout", "Interval", "ApprovalDelay",
 
762
                               "ApprovalDuration", "ExtendedTimeout"):
 
763
                    return cls.milliseconds_to_string(value)
 
764
                return str(value)
 
765
 
 
766
            def header_line(self, format_string):
 
767
                return format_string.format(**self.tableheaders)
 
768
 
 
769
            def client_line(self, client, format_string):
 
770
                return format_string.format(
 
771
                    **{key: self.string_from_client(client, key)
 
772
                       for key in self.keywords})
 
773
 
 
774
            @staticmethod
 
775
            def milliseconds_to_string(ms):
 
776
                td = datetime.timedelta(0, 0, 0, ms)
 
777
                return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
 
778
                        .format(days="{}T".format(td.days)
 
779
                                if td.days else "",
 
780
                                hours=td.seconds // 3600,
 
781
                                minutes=(td.seconds % 3600) // 60,
 
782
                                seconds=td.seconds % 60))
 
783
 
 
784
 
 
785
    class PropertySetter(Base):
 
786
        "Abstract class for Actions for setting one client property"
 
787
 
 
788
        def run_on_one_client(self, client, properties=None):
 
789
            """Set the Client's D-Bus property"""
 
790
            self.bus.set_client_property(client, self.propname,
 
791
                                         self.value_to_set)
 
792
 
 
793
        @property
 
794
        def propname(self):
 
795
            raise NotImplementedError()
 
796
 
 
797
 
 
798
    class Enable(PropertySetter):
 
799
        propname = "Enabled"
 
800
        value_to_set = True
 
801
 
 
802
 
 
803
    class Disable(PropertySetter):
 
804
        propname = "Enabled"
 
805
        value_to_set = False
 
806
 
 
807
 
 
808
    class BumpTimeout(PropertySetter):
 
809
        propname = "LastCheckedOK"
 
810
        value_to_set = ""
 
811
 
 
812
 
 
813
    class StartChecker(PropertySetter):
 
814
        propname = "CheckerRunning"
 
815
        value_to_set = True
 
816
 
 
817
 
 
818
    class StopChecker(PropertySetter):
 
819
        propname = "CheckerRunning"
 
820
        value_to_set = False
 
821
 
 
822
 
 
823
    class ApproveByDefault(PropertySetter):
 
824
        propname = "ApprovedByDefault"
 
825
        value_to_set = True
 
826
 
 
827
 
 
828
    class DenyByDefault(PropertySetter):
 
829
        propname = "ApprovedByDefault"
 
830
        value_to_set = False
 
831
 
 
832
 
 
833
    class PropertySetterValue(PropertySetter):
 
834
        """Abstract class for PropertySetter recieving a value as
 
835
constructor argument instead of a class attribute."""
 
836
        def __init__(self, value):
 
837
            self.value_to_set = value
 
838
 
 
839
        @classmethod
 
840
        def argparse(cls, argtype):
 
841
            def cmdtype(arg):
 
842
                return cls(argtype(arg))
 
843
            return cmdtype
 
844
 
 
845
    class SetChecker(PropertySetterValue):
 
846
        propname = "Checker"
 
847
 
 
848
 
 
849
    class SetHost(PropertySetterValue):
 
850
        propname = "Host"
 
851
 
 
852
 
 
853
    class SetSecret(PropertySetterValue):
 
854
        propname = "Secret"
 
855
 
 
856
        @property
 
857
        def value_to_set(self):
 
858
            return self._vts
 
859
 
 
860
        @value_to_set.setter
 
861
        def value_to_set(self, value):
 
862
            """When setting, read data from supplied file object"""
 
863
            self._vts = value.read()
 
864
            value.close()
 
865
 
 
866
 
 
867
    class PropertySetterValueMilliseconds(PropertySetterValue):
 
868
        """Abstract class for PropertySetterValue taking a value
 
869
argument as a datetime.timedelta() but should store it as
 
870
milliseconds."""
 
871
 
 
872
        @property
 
873
        def value_to_set(self):
 
874
            return self._vts
 
875
 
 
876
        @value_to_set.setter
 
877
        def value_to_set(self, value):
 
878
            "When setting, convert value from a datetime.timedelta"
 
879
            self._vts = int(round(value.total_seconds() * 1000))
 
880
 
 
881
 
 
882
    class SetTimeout(PropertySetterValueMilliseconds):
 
883
        propname = "Timeout"
 
884
 
 
885
 
 
886
    class SetExtendedTimeout(PropertySetterValueMilliseconds):
 
887
        propname = "ExtendedTimeout"
 
888
 
 
889
 
 
890
    class SetInterval(PropertySetterValueMilliseconds):
 
891
        propname = "Interval"
 
892
 
 
893
 
 
894
    class SetApprovalDelay(PropertySetterValueMilliseconds):
 
895
        propname = "ApprovalDelay"
 
896
 
 
897
 
 
898
    class SetApprovalDuration(PropertySetterValueMilliseconds):
 
899
        propname = "ApprovalDuration"
 
900
 
 
901
 
 
902
 
 
903
class TestCaseWithAssertLogs(unittest.TestCase):
 
904
    """unittest.TestCase.assertLogs only exists in Python 3.4"""
 
905
 
 
906
    if not hasattr(unittest.TestCase, "assertLogs"):
 
907
        @contextlib.contextmanager
 
908
        def assertLogs(self, logger, level=logging.INFO):
 
909
            capturing_handler = self.CapturingLevelHandler(level)
 
910
            old_level = logger.level
 
911
            old_propagate = logger.propagate
 
912
            logger.addHandler(capturing_handler)
 
913
            logger.setLevel(level)
 
914
            logger.propagate = False
 
915
            try:
 
916
                yield capturing_handler.watcher
 
917
            finally:
 
918
                logger.propagate = old_propagate
 
919
                logger.removeHandler(capturing_handler)
 
920
                logger.setLevel(old_level)
 
921
            self.assertGreater(len(capturing_handler.watcher.records),
 
922
                               0)
 
923
 
 
924
        class CapturingLevelHandler(logging.Handler):
 
925
            def __init__(self, level, *args, **kwargs):
 
926
                logging.Handler.__init__(self, *args, **kwargs)
 
927
                self.watcher = self.LoggingWatcher([], [])
 
928
            def emit(self, record):
 
929
                self.watcher.records.append(record)
 
930
                self.watcher.output.append(self.format(record))
 
931
 
 
932
            LoggingWatcher = collections.namedtuple("LoggingWatcher",
 
933
                                                    ("records",
 
934
                                                     "output"))
 
935
 
 
936
 
 
937
class Unique(object):
 
938
    """Class for objects which exist only to be unique objects, since
 
939
unittest.mock.sentinel only exists in Python 3.3"""
 
940
 
 
941
 
 
942
class Test_string_to_delta(TestCaseWithAssertLogs):
 
943
    # Just test basic RFC 3339 functionality here, the doc string for
 
944
    # rfc3339_duration_to_delta() already has more comprehensive
 
945
    # tests, which are run by doctest.
 
946
 
 
947
    def test_rfc3339_zero_seconds(self):
 
948
        self.assertEqual(datetime.timedelta(),
 
949
                         string_to_delta("PT0S"))
 
950
 
 
951
    def test_rfc3339_zero_days(self):
 
952
        self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
 
953
 
 
954
    def test_rfc3339_one_second(self):
 
955
        self.assertEqual(datetime.timedelta(0, 1),
 
956
                         string_to_delta("PT1S"))
 
957
 
 
958
    def test_rfc3339_two_hours(self):
 
959
        self.assertEqual(datetime.timedelta(0, 7200),
 
960
                         string_to_delta("PT2H"))
 
961
 
 
962
    def test_falls_back_to_pre_1_6_1_with_warning(self):
 
963
        with self.assertLogs(log, logging.WARNING):
 
964
            value = string_to_delta("2h")
 
965
        self.assertEqual(datetime.timedelta(0, 7200), value)
 
966
 
 
967
 
 
968
class Test_check_option_syntax(unittest.TestCase):
 
969
    def setUp(self):
 
970
        self.parser = argparse.ArgumentParser()
 
971
        add_command_line_options(self.parser)
 
972
 
 
973
    def test_actions_requires_client_or_all(self):
 
974
        for action, value in self.actions.items():
 
975
            args = self.actionargs(action, value)
 
976
            with self.assertParseError():
 
977
                self.parse_args(args)
 
978
 
 
979
    # This mostly corresponds to the definition from has_commands() in
 
980
    # check_option_syntax()
 
981
    actions = {
 
982
        "--enable": None,
 
983
        "--disable": None,
 
984
        "--bump-timeout": None,
 
985
        "--start-checker": None,
 
986
        "--stop-checker": None,
 
987
        "--is-enabled": None,
 
988
        "--remove": None,
 
989
        "--checker": "x",
 
990
        "--timeout": "PT0S",
 
991
        "--extended-timeout": "PT0S",
 
992
        "--interval": "PT0S",
 
993
        "--approve-by-default": None,
 
994
        "--deny-by-default": None,
 
995
        "--approval-delay": "PT0S",
 
996
        "--approval-duration": "PT0S",
 
997
        "--host": "hostname",
 
998
        "--secret": "/dev/null",
 
999
        "--approve": None,
 
1000
        "--deny": None,
 
1001
    }
 
1002
 
 
1003
    @staticmethod
 
1004
    def actionargs(action, value, *args):
 
1005
        if value is not None:
 
1006
            return [action, value] + list(args)
 
1007
        else:
 
1008
            return [action] + list(args)
 
1009
 
 
1010
    @contextlib.contextmanager
 
1011
    def assertParseError(self):
 
1012
        with self.assertRaises(SystemExit) as e:
 
1013
            with self.redirect_stderr_to_devnull():
 
1014
                yield
 
1015
        # Exit code from argparse is guaranteed to be "2".  Reference:
 
1016
        # https://docs.python.org/3/library
 
1017
        # /argparse.html#exiting-methods
 
1018
        self.assertEqual(2, e.exception.code)
 
1019
 
 
1020
    def parse_args(self, args):
 
1021
        options = self.parser.parse_args(args)
 
1022
        check_option_syntax(self.parser, options)
 
1023
 
 
1024
    @staticmethod
 
1025
    @contextlib.contextmanager
 
1026
    def redirect_stderr_to_devnull():
 
1027
        old_stderr = sys.stderr
 
1028
        with contextlib.closing(open(os.devnull, "w")) as null:
 
1029
            sys.stderr = null
 
1030
            try:
 
1031
                yield
 
1032
            finally:
 
1033
                sys.stderr = old_stderr
 
1034
 
 
1035
    def check_option_syntax(self, options):
 
1036
        check_option_syntax(self.parser, options)
 
1037
 
 
1038
    def test_actions_all_conflicts_with_verbose(self):
 
1039
        for action, value in self.actions.items():
 
1040
            args = self.actionargs(action, value, "--all",
 
1041
                                   "--verbose")
 
1042
            with self.assertParseError():
 
1043
                self.parse_args(args)
 
1044
 
 
1045
    def test_actions_with_client_conflicts_with_verbose(self):
 
1046
        for action, value in self.actions.items():
 
1047
            args = self.actionargs(action, value, "--verbose",
 
1048
                                   "client")
 
1049
            with self.assertParseError():
 
1050
                self.parse_args(args)
 
1051
 
 
1052
    def test_dump_json_conflicts_with_verbose(self):
 
1053
        args = ["--dump-json", "--verbose"]
 
1054
        with self.assertParseError():
 
1055
            self.parse_args(args)
 
1056
 
 
1057
    def test_dump_json_conflicts_with_action(self):
 
1058
        for action, value in self.actions.items():
 
1059
            args = self.actionargs(action, value, "--dump-json")
 
1060
            with self.assertParseError():
 
1061
                self.parse_args(args)
 
1062
 
 
1063
    def test_all_can_not_be_alone(self):
 
1064
        args = ["--all"]
 
1065
        with self.assertParseError():
 
1066
            self.parse_args(args)
 
1067
 
 
1068
    def test_all_is_ok_with_any_action(self):
 
1069
        for action, value in self.actions.items():
 
1070
            args = self.actionargs(action, value, "--all")
 
1071
            self.parse_args(args)
 
1072
 
 
1073
    def test_any_action_is_ok_with_one_client(self):
 
1074
        for action, value in self.actions.items():
 
1075
            args = self.actionargs(action, value, "client")
 
1076
            self.parse_args(args)
 
1077
 
 
1078
    def test_one_client_with_all_actions_except_is_enabled(self):
 
1079
        for action, value in self.actions.items():
 
1080
            if action == "--is-enabled":
 
1081
                continue
 
1082
            args = self.actionargs(action, value, "client")
 
1083
            self.parse_args(args)
 
1084
 
 
1085
    def test_two_clients_with_all_actions_except_is_enabled(self):
 
1086
        for action, value in self.actions.items():
 
1087
            if action == "--is-enabled":
 
1088
                continue
 
1089
            args = self.actionargs(action, value, "client1",
 
1090
                                   "client2")
 
1091
            self.parse_args(args)
 
1092
 
 
1093
    def test_two_clients_are_ok_with_actions_except_is_enabled(self):
 
1094
        for action, value in self.actions.items():
 
1095
            if action == "--is-enabled":
 
1096
                continue
 
1097
            args = self.actionargs(action, value, "client1",
 
1098
                                   "client2")
 
1099
            self.parse_args(args)
 
1100
 
 
1101
    def test_is_enabled_fails_without_client(self):
 
1102
        args = ["--is-enabled"]
 
1103
        with self.assertParseError():
 
1104
            self.parse_args(args)
 
1105
 
 
1106
    def test_is_enabled_fails_with_two_clients(self):
 
1107
        args = ["--is-enabled", "client1", "client2"]
 
1108
        with self.assertParseError():
 
1109
            self.parse_args(args)
 
1110
 
 
1111
    def test_remove_can_only_be_combined_with_action_deny(self):
 
1112
        for action, value in self.actions.items():
 
1113
            if action in {"--remove", "--deny"}:
 
1114
                continue
 
1115
            args = self.actionargs(action, value, "--all",
 
1116
                                   "--remove")
 
1117
            with self.assertParseError():
 
1118
                self.parse_args(args)
 
1119
 
 
1120
 
 
1121
class Test_dbus_exceptions(unittest.TestCase):
 
1122
 
 
1123
    def test_dbus_ConnectFailed_is_Error(self):
 
1124
        with self.assertRaises(dbus.Error):
 
1125
            raise dbus.ConnectFailed()
 
1126
 
 
1127
 
 
1128
class Test_dbus_MandosBus(unittest.TestCase):
 
1129
 
 
1130
    class MockMandosBus(dbus.MandosBus):
 
1131
        def __init__(self):
 
1132
            self._name = "se.recompile.Mandos"
 
1133
            self._server_path = "/"
 
1134
            self._server_interface = "se.recompile.Mandos"
 
1135
            self._client_interface = "se.recompile.Mandos.Client"
 
1136
            self.calls = []
 
1137
            self.call_method_return = Unique()
 
1138
 
 
1139
        def call_method(self, methodname, busname, objectpath,
 
1140
                        interface, *args):
 
1141
            self.calls.append((methodname, busname, objectpath,
 
1142
                               interface, args))
 
1143
            return self.call_method_return
 
1144
 
 
1145
    def setUp(self):
 
1146
        self.bus = self.MockMandosBus()
 
1147
 
 
1148
    def test_set_client_property(self):
 
1149
        self.bus.set_client_property("objectpath", "key", "value")
 
1150
        expected_call = ("Set", self.bus._name, "objectpath",
 
1151
                         "org.freedesktop.DBus.Properties",
 
1152
                         (self.bus._client_interface, "key", "value"))
 
1153
        self.assertIn(expected_call, self.bus.calls)
 
1154
 
 
1155
    def test_call_client_method(self):
 
1156
        ret = self.bus.call_client_method("objectpath", "methodname")
 
1157
        self.assertIs(self.bus.call_method_return, ret)
 
1158
        expected_call = ("methodname", self.bus._name, "objectpath",
 
1159
                         self.bus._client_interface, ())
 
1160
        self.assertIn(expected_call, self.bus.calls)
 
1161
 
 
1162
    def test_call_client_method_with_args(self):
 
1163
        args = (Unique(), Unique())
 
1164
        ret = self.bus.call_client_method("objectpath", "methodname",
 
1165
                                          *args)
 
1166
        self.assertIs(self.bus.call_method_return, ret)
 
1167
        expected_call = ("methodname", self.bus._name, "objectpath",
 
1168
                         self.bus._client_interface,
 
1169
                         (args[0], args[1]))
 
1170
        self.assertIn(expected_call, self.bus.calls)
 
1171
 
 
1172
    def test_get_clients_and_properties(self):
 
1173
        managed_objects = {
 
1174
            "objectpath": {
 
1175
                self.bus._client_interface: {
 
1176
                    "key": "value",
 
1177
                    "bool": True,
 
1178
                },
 
1179
                "irrelevant_interface": {
 
1180
                    "key": "othervalue",
 
1181
                    "bool": False,
 
1182
                },
 
1183
            },
 
1184
            "other_objectpath": {
 
1185
                "other_irrelevant_interface": {
 
1186
                    "key": "value 3",
 
1187
                    "bool": None,
 
1188
                },
 
1189
            },
 
1190
        }
 
1191
        expected_clients_and_properties = {
 
1192
            "objectpath": {
 
1193
                "key": "value",
 
1194
                "bool": True,
 
1195
            }
 
1196
        }
 
1197
        self.bus.call_method_return = managed_objects
 
1198
        ret = self.bus.get_clients_and_properties()
 
1199
        self.assertDictEqual(expected_clients_and_properties, ret)
 
1200
        expected_call = ("GetManagedObjects", self.bus._name,
 
1201
                         self.bus._server_path,
 
1202
                         "org.freedesktop.DBus.ObjectManager", ())
 
1203
        self.assertIn(expected_call, self.bus.calls)
 
1204
 
 
1205
    def test_call_server_method(self):
 
1206
        ret = self.bus.call_server_method("methodname")
 
1207
        self.assertIs(self.bus.call_method_return, ret)
 
1208
        expected_call = ("methodname", self.bus._name,
 
1209
                         self.bus._server_path,
 
1210
                         self.bus._server_interface, ())
 
1211
        self.assertIn(expected_call, self.bus.calls)
 
1212
 
 
1213
    def test_call_server_method_with_args(self):
 
1214
        args = (Unique(), Unique())
 
1215
        ret = self.bus.call_server_method("methodname", *args)
 
1216
        self.assertIs(self.bus.call_method_return, ret)
 
1217
        expected_call = ("methodname", self.bus._name,
 
1218
                         self.bus._server_path,
 
1219
                         self.bus._server_interface,
 
1220
                         (args[0], args[1]))
 
1221
        self.assertIn(expected_call, self.bus.calls)
 
1222
 
 
1223
 
 
1224
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
 
1225
 
 
1226
    def MockDBusPython_func(self, func):
 
1227
        class mock_dbus_python(object):
 
1228
            """mock dbus-python module"""
 
1229
            class exceptions(object):
 
1230
                """Pseudo-namespace"""
 
1231
                class DBusException(Exception):
 
1232
                    pass
 
1233
            class SystemBus(object):
 
1234
                @staticmethod
 
1235
                def get_object(busname, objectpath):
 
1236
                    DBusObject = collections.namedtuple(
 
1237
                        "DBusObject", ("methodname",))
 
1238
                    def method(*args, **kwargs):
 
1239
                        self.assertEqual({"dbus_interface":
 
1240
                                          "interface"},
 
1241
                                         kwargs)
 
1242
                        return func(*args)
 
1243
                    return DBusObject(methodname=method)
 
1244
            class Boolean(object):
 
1245
                def __init__(self, value):
 
1246
                    self.value = bool(value)
 
1247
                def __bool__(self):
 
1248
                    return self.value
 
1249
                if sys.version_info.major == 2:
 
1250
                    __nonzero__ = __bool__
 
1251
            class ObjectPath(str):
 
1252
                pass
 
1253
            class Dictionary(dict):
 
1254
                pass
 
1255
        return mock_dbus_python
 
1256
 
 
1257
    def call_method(self, bus, methodname, busname, objectpath,
 
1258
                    interface, *args):
 
1259
        with self.assertLogs(log, logging.DEBUG):
 
1260
            return bus.call_method(methodname, busname, objectpath,
 
1261
                                   interface, *args)
 
1262
 
 
1263
    def test_call_method_returns(self):
 
1264
        expected_method_return = Unique()
 
1265
        method_args = (Unique(), Unique())
 
1266
        def func(*args):
 
1267
            self.assertEqual(len(method_args), len(args))
 
1268
            for marg, arg in zip(method_args, args):
 
1269
                self.assertIs(marg, arg)
 
1270
            return expected_method_return
 
1271
        mock_dbus_python = self.MockDBusPython_func(func)
 
1272
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1273
        ret = self.call_method(bus, "methodname", "busname",
 
1274
                               "objectpath", "interface",
 
1275
                               *method_args)
 
1276
        self.assertIs(ret, expected_method_return)
 
1277
 
 
1278
    def test_call_method_filters_bool_true(self):
 
1279
        def func():
 
1280
            return method_return
 
1281
        mock_dbus_python = self.MockDBusPython_func(func)
 
1282
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1283
        method_return = mock_dbus_python.Boolean(True)
 
1284
        ret = self.call_method(bus, "methodname", "busname",
 
1285
                               "objectpath", "interface")
 
1286
        self.assertTrue(ret)
 
1287
        self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
 
1288
 
 
1289
    def test_call_method_filters_bool_false(self):
 
1290
        def func():
 
1291
            return method_return
 
1292
        mock_dbus_python = self.MockDBusPython_func(func)
 
1293
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1294
        method_return = mock_dbus_python.Boolean(False)
 
1295
        ret = self.call_method(bus, "methodname", "busname",
 
1296
                               "objectpath", "interface")
 
1297
        self.assertFalse(ret)
 
1298
        self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
 
1299
 
 
1300
    def test_call_method_filters_objectpath(self):
 
1301
        def func():
 
1302
            return method_return
 
1303
        mock_dbus_python = self.MockDBusPython_func(func)
 
1304
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1305
        method_return = mock_dbus_python.ObjectPath("objectpath")
 
1306
        ret = self.call_method(bus, "methodname", "busname",
 
1307
                               "objectpath", "interface")
 
1308
        self.assertEqual("objectpath", ret)
 
1309
        self.assertIsNot("objectpath", ret)
 
1310
        self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
 
1311
 
 
1312
    def test_call_method_filters_booleans_in_dict(self):
 
1313
        def func():
 
1314
            return method_return
 
1315
        mock_dbus_python = self.MockDBusPython_func(func)
 
1316
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1317
        method_return = mock_dbus_python.Dictionary(
 
1318
        {mock_dbus_python.Boolean(True):
 
1319
         mock_dbus_python.Boolean(False),
 
1320
         mock_dbus_python.Boolean(False):
 
1321
         mock_dbus_python.Boolean(True)})
 
1322
        ret = self.call_method(bus, "methodname", "busname",
 
1323
                               "objectpath", "interface")
 
1324
        expected_method_return = {True: False,
 
1325
                                  False: True}
 
1326
        self.assertEqual(expected_method_return, ret)
 
1327
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1328
 
 
1329
    def test_call_method_filters_objectpaths_in_dict(self):
 
1330
        def func():
 
1331
            return method_return
 
1332
        mock_dbus_python = self.MockDBusPython_func(func)
 
1333
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1334
        method_return = mock_dbus_python.Dictionary(
 
1335
        {mock_dbus_python.ObjectPath("objectpath_key_1"):
 
1336
         mock_dbus_python.ObjectPath("objectpath_value_1"),
 
1337
         mock_dbus_python.ObjectPath("objectpath_key_2"):
 
1338
         mock_dbus_python.ObjectPath("objectpath_value_2")})
 
1339
        ret = self.call_method(bus, "methodname", "busname",
 
1340
                               "objectpath", "interface")
 
1341
        expected_method_return = {str(key): str(value)
 
1342
                                  for key, value in
 
1343
                                  method_return.items()}
 
1344
        self.assertEqual(expected_method_return, ret)
 
1345
        self.assertIsInstance(ret, dict)
 
1346
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1347
 
 
1348
    def test_call_method_filters_dict_in_dict(self):
 
1349
        def func():
 
1350
            return method_return
 
1351
        mock_dbus_python = self.MockDBusPython_func(func)
 
1352
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1353
        method_return = mock_dbus_python.Dictionary(
 
1354
        {"key1": mock_dbus_python.Dictionary({"key11": "value11",
 
1355
                                              "key12": "value12"}),
 
1356
         "key2": mock_dbus_python.Dictionary({"key21": "value21",
 
1357
                                              "key22": "value22"})})
 
1358
        ret = self.call_method(bus, "methodname", "busname",
 
1359
                               "objectpath", "interface")
 
1360
        expected_method_return = {
 
1361
            "key1": {"key11": "value11",
 
1362
                     "key12": "value12"},
 
1363
            "key2": {"key21": "value21",
 
1364
                     "key22": "value22"},
 
1365
        }
 
1366
        self.assertEqual(expected_method_return, ret)
 
1367
        self.assertIsInstance(ret, dict)
 
1368
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1369
        for key, value in ret.items():
 
1370
            self.assertIsInstance(value, dict)
 
1371
            self.assertEqual(expected_method_return[key], value)
 
1372
            self.assertNotIsInstance(value,
 
1373
                                     mock_dbus_python.Dictionary)
 
1374
 
 
1375
    def test_call_method_filters_dict_three_deep(self):
 
1376
        def func():
 
1377
            return method_return
 
1378
        mock_dbus_python = self.MockDBusPython_func(func)
 
1379
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1380
        method_return = mock_dbus_python.Dictionary(
 
1381
            {"key1":
 
1382
             mock_dbus_python.Dictionary(
 
1383
                 {"key2":
 
1384
                  mock_dbus_python.Dictionary(
 
1385
                      {"key3":
 
1386
                       mock_dbus_python.Boolean(True),
 
1387
                       }),
 
1388
                  }),
 
1389
             })
 
1390
        ret = self.call_method(bus, "methodname", "busname",
 
1391
                               "objectpath", "interface")
 
1392
        expected_method_return = {"key1": {"key2": {"key3": True}}}
 
1393
        self.assertEqual(expected_method_return, ret)
 
1394
        self.assertIsInstance(ret, dict)
 
1395
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1396
        self.assertIsInstance(ret["key1"], dict)
 
1397
        self.assertNotIsInstance(ret["key1"],
 
1398
                                 mock_dbus_python.Dictionary)
 
1399
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
1400
        self.assertNotIsInstance(ret["key1"]["key2"],
 
1401
                                 mock_dbus_python.Dictionary)
 
1402
        self.assertTrue(ret["key1"]["key2"]["key3"])
 
1403
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
1404
                                 mock_dbus_python.Boolean)
 
1405
 
 
1406
    def test_call_method_handles_exception(self):
 
1407
        dbus_logger = logging.getLogger("dbus.proxies")
 
1408
 
 
1409
        def func():
 
1410
            dbus_logger.error("Test")
 
1411
            raise mock_dbus_python.exceptions.DBusException()
 
1412
 
 
1413
        mock_dbus_python = self.MockDBusPython_func(func)
 
1414
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1415
 
 
1416
        class CountingHandler(logging.Handler):
 
1417
            count = 0
 
1418
            def emit(self, record):
 
1419
                self.count += 1
 
1420
 
 
1421
        counting_handler = CountingHandler()
 
1422
 
 
1423
        dbus_logger.addHandler(counting_handler)
 
1424
 
 
1425
        try:
 
1426
            with self.assertRaises(dbus.Error) as e:
 
1427
                self.call_method(bus, "methodname", "busname",
 
1428
                                 "objectpath", "interface")
 
1429
        finally:
 
1430
            dbus_logger.removeFilter(counting_handler)
 
1431
 
 
1432
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1433
 
 
1434
        # Make sure the dbus logger was suppressed
 
1435
        self.assertEqual(0, counting_handler.count)
 
1436
 
 
1437
    def test_get_object_converts_to_correct_exception(self):
 
1438
        bus = dbus_python_adapter.SystemBus(
 
1439
            self.fake_dbus_python_raises_exception_on_connect)
 
1440
        with self.assertRaises(dbus.ConnectFailed):
 
1441
            self.call_method(bus, "methodname", "busname",
 
1442
                             "objectpath", "interface")
 
1443
 
 
1444
    class fake_dbus_python_raises_exception_on_connect(object):
 
1445
        """fake dbus-python module"""
 
1446
        class exceptions(object):
 
1447
            """Pseudo-namespace"""
 
1448
            class DBusException(Exception):
 
1449
                pass
 
1450
 
 
1451
        @classmethod
 
1452
        def SystemBus(cls):
 
1453
            def get_object(busname, objectpath):
 
1454
                raise cls.exceptions.DBusException()
 
1455
            Bus = collections.namedtuple("Bus", ["get_object"])
 
1456
            return Bus(get_object=get_object)
 
1457
 
 
1458
 
 
1459
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
 
1460
    class mock_dbus_python(object):
 
1461
        """mock dbus-python modules"""
 
1462
        class SystemBus(object):
 
1463
            @staticmethod
 
1464
            def get_object(busname, objectpath):
 
1465
                return Unique()
 
1466
 
 
1467
    def setUp(self):
 
1468
        self.bus = dbus_python_adapter.CachingBus(
 
1469
            self.mock_dbus_python)
 
1470
 
 
1471
    def test_returns_distinct_objectpaths(self):
 
1472
        obj1 = self.bus.get_object("busname", "objectpath1")
 
1473
        self.assertIsInstance(obj1, Unique)
 
1474
        obj2 = self.bus.get_object("busname", "objectpath2")
 
1475
        self.assertIsInstance(obj2, Unique)
 
1476
        self.assertIsNot(obj1, obj2)
 
1477
 
 
1478
    def test_returns_distinct_busnames(self):
 
1479
        obj1 = self.bus.get_object("busname1", "objectpath")
 
1480
        self.assertIsInstance(obj1, Unique)
 
1481
        obj2 = self.bus.get_object("busname2", "objectpath")
 
1482
        self.assertIsInstance(obj2, Unique)
 
1483
        self.assertIsNot(obj1, obj2)
 
1484
 
 
1485
    def test_returns_distinct_both(self):
 
1486
        obj1 = self.bus.get_object("busname1", "objectpath")
 
1487
        self.assertIsInstance(obj1, Unique)
 
1488
        obj2 = self.bus.get_object("busname2", "objectpath")
 
1489
        self.assertIsInstance(obj2, Unique)
 
1490
        self.assertIsNot(obj1, obj2)
 
1491
 
 
1492
    def test_returns_same(self):
 
1493
        obj1 = self.bus.get_object("busname", "objectpath")
 
1494
        self.assertIsInstance(obj1, Unique)
 
1495
        obj2 = self.bus.get_object("busname", "objectpath")
 
1496
        self.assertIsInstance(obj2, Unique)
 
1497
        self.assertIs(obj1, obj2)
 
1498
 
 
1499
    def test_returns_same_old(self):
 
1500
        obj1 = self.bus.get_object("busname1", "objectpath1")
 
1501
        self.assertIsInstance(obj1, Unique)
 
1502
        obj2 = self.bus.get_object("busname2", "objectpath2")
 
1503
        self.assertIsInstance(obj2, Unique)
 
1504
        obj1b = self.bus.get_object("busname1", "objectpath1")
 
1505
        self.assertIsInstance(obj1b, Unique)
 
1506
        self.assertIsNot(obj1, obj2)
 
1507
        self.assertIsNot(obj2, obj1b)
 
1508
        self.assertIs(obj1, obj1b)
 
1509
 
 
1510
 
 
1511
class Test_commands_from_options(unittest.TestCase):
 
1512
 
 
1513
    def setUp(self):
 
1514
        self.parser = argparse.ArgumentParser()
 
1515
        add_command_line_options(self.parser)
 
1516
 
 
1517
    def test_is_enabled(self):
 
1518
        self.assert_command_from_args(["--is-enabled", "client"],
 
1519
                                      command.IsEnabled)
 
1520
 
 
1521
    def assert_command_from_args(self, args, command_cls,
 
1522
                                 **cmd_attrs):
 
1523
        """Assert that parsing ARGS should result in an instance of
 
1524
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
 
1525
        options = self.parser.parse_args(args)
 
1526
        check_option_syntax(self.parser, options)
 
1527
        commands = commands_from_options(options)
 
1528
        self.assertEqual(1, len(commands))
 
1529
        command = commands[0]
 
1530
        self.assertIsInstance(command, command_cls)
 
1531
        for key, value in cmd_attrs.items():
 
1532
            self.assertEqual(value, getattr(command, key))
 
1533
 
 
1534
    def test_is_enabled_short(self):
 
1535
        self.assert_command_from_args(["-V", "client"],
 
1536
                                      command.IsEnabled)
 
1537
 
 
1538
    def test_approve(self):
 
1539
        self.assert_command_from_args(["--approve", "client"],
 
1540
                                      command.Approve)
 
1541
 
 
1542
    def test_approve_short(self):
 
1543
        self.assert_command_from_args(["-A", "client"],
 
1544
                                      command.Approve)
 
1545
 
 
1546
    def test_deny(self):
 
1547
        self.assert_command_from_args(["--deny", "client"],
 
1548
                                      command.Deny)
 
1549
 
 
1550
    def test_deny_short(self):
 
1551
        self.assert_command_from_args(["-D", "client"], command.Deny)
 
1552
 
 
1553
    def test_remove(self):
 
1554
        self.assert_command_from_args(["--remove", "client"],
 
1555
                                      command.Remove)
 
1556
 
 
1557
    def test_deny_before_remove(self):
 
1558
        options = self.parser.parse_args(["--deny", "--remove",
 
1559
                                          "client"])
 
1560
        check_option_syntax(self.parser, options)
 
1561
        commands = commands_from_options(options)
 
1562
        self.assertEqual(2, len(commands))
 
1563
        self.assertIsInstance(commands[0], command.Deny)
 
1564
        self.assertIsInstance(commands[1], command.Remove)
 
1565
 
 
1566
    def test_deny_before_remove_reversed(self):
 
1567
        options = self.parser.parse_args(["--remove", "--deny",
 
1568
                                          "--all"])
 
1569
        check_option_syntax(self.parser, options)
 
1570
        commands = commands_from_options(options)
 
1571
        self.assertEqual(2, len(commands))
 
1572
        self.assertIsInstance(commands[0], command.Deny)
 
1573
        self.assertIsInstance(commands[1], command.Remove)
 
1574
 
 
1575
    def test_remove_short(self):
 
1576
        self.assert_command_from_args(["-r", "client"],
 
1577
                                      command.Remove)
 
1578
 
 
1579
    def test_dump_json(self):
 
1580
        self.assert_command_from_args(["--dump-json"],
 
1581
                                      command.DumpJSON)
 
1582
 
 
1583
    def test_enable(self):
 
1584
        self.assert_command_from_args(["--enable", "client"],
 
1585
                                      command.Enable)
 
1586
 
 
1587
    def test_enable_short(self):
 
1588
        self.assert_command_from_args(["-e", "client"],
 
1589
                                      command.Enable)
 
1590
 
 
1591
    def test_disable(self):
 
1592
        self.assert_command_from_args(["--disable", "client"],
 
1593
                                      command.Disable)
 
1594
 
 
1595
    def test_disable_short(self):
 
1596
        self.assert_command_from_args(["-d", "client"],
 
1597
                                      command.Disable)
 
1598
 
 
1599
    def test_bump_timeout(self):
 
1600
        self.assert_command_from_args(["--bump-timeout", "client"],
 
1601
                                      command.BumpTimeout)
 
1602
 
 
1603
    def test_bump_timeout_short(self):
 
1604
        self.assert_command_from_args(["-b", "client"],
 
1605
                                      command.BumpTimeout)
 
1606
 
 
1607
    def test_start_checker(self):
 
1608
        self.assert_command_from_args(["--start-checker", "client"],
 
1609
                                      command.StartChecker)
 
1610
 
 
1611
    def test_stop_checker(self):
 
1612
        self.assert_command_from_args(["--stop-checker", "client"],
 
1613
                                      command.StopChecker)
 
1614
 
 
1615
    def test_approve_by_default(self):
 
1616
        self.assert_command_from_args(["--approve-by-default",
 
1617
                                       "client"],
 
1618
                                      command.ApproveByDefault)
 
1619
 
 
1620
    def test_deny_by_default(self):
 
1621
        self.assert_command_from_args(["--deny-by-default", "client"],
 
1622
                                      command.DenyByDefault)
 
1623
 
 
1624
    def test_checker(self):
 
1625
        self.assert_command_from_args(["--checker", ":", "client"],
 
1626
                                      command.SetChecker,
 
1627
                                      value_to_set=":")
 
1628
 
 
1629
    def test_checker_empty(self):
 
1630
        self.assert_command_from_args(["--checker", "", "client"],
 
1631
                                      command.SetChecker,
 
1632
                                      value_to_set="")
 
1633
 
 
1634
    def test_checker_short(self):
 
1635
        self.assert_command_from_args(["-c", ":", "client"],
 
1636
                                      command.SetChecker,
 
1637
                                      value_to_set=":")
 
1638
 
 
1639
    def test_host(self):
 
1640
        self.assert_command_from_args(
 
1641
            ["--host", "client.example.org", "client"],
 
1642
            command.SetHost, value_to_set="client.example.org")
 
1643
 
 
1644
    def test_host_short(self):
 
1645
        self.assert_command_from_args(
 
1646
            ["-H", "client.example.org", "client"], command.SetHost,
 
1647
            value_to_set="client.example.org")
 
1648
 
 
1649
    def test_secret_devnull(self):
 
1650
        self.assert_command_from_args(["--secret", os.path.devnull,
 
1651
                                       "client"], command.SetSecret,
 
1652
                                      value_to_set=b"")
 
1653
 
 
1654
    def test_secret_tempfile(self):
 
1655
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
 
1656
            value = b"secret\0xyzzy\nbar"
 
1657
            f.write(value)
 
1658
            f.seek(0)
 
1659
            self.assert_command_from_args(["--secret", f.name,
 
1660
                                           "client"],
 
1661
                                          command.SetSecret,
 
1662
                                          value_to_set=value)
 
1663
 
 
1664
    def test_secret_devnull_short(self):
 
1665
        self.assert_command_from_args(["-s", os.path.devnull,
 
1666
                                       "client"], command.SetSecret,
 
1667
                                      value_to_set=b"")
 
1668
 
 
1669
    def test_secret_tempfile_short(self):
 
1670
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
 
1671
            value = b"secret\0xyzzy\nbar"
 
1672
            f.write(value)
 
1673
            f.seek(0)
 
1674
            self.assert_command_from_args(["-s", f.name, "client"],
 
1675
                                          command.SetSecret,
 
1676
                                          value_to_set=value)
 
1677
 
 
1678
    def test_timeout(self):
 
1679
        self.assert_command_from_args(["--timeout", "PT5M", "client"],
 
1680
                                      command.SetTimeout,
 
1681
                                      value_to_set=300000)
 
1682
 
 
1683
    def test_timeout_short(self):
 
1684
        self.assert_command_from_args(["-t", "PT5M", "client"],
 
1685
                                      command.SetTimeout,
 
1686
                                      value_to_set=300000)
 
1687
 
 
1688
    def test_extended_timeout(self):
 
1689
        self.assert_command_from_args(["--extended-timeout", "PT15M",
 
1690
                                       "client"],
 
1691
                                      command.SetExtendedTimeout,
 
1692
                                      value_to_set=900000)
 
1693
 
 
1694
    def test_interval(self):
 
1695
        self.assert_command_from_args(["--interval", "PT2M",
 
1696
                                       "client"], command.SetInterval,
 
1697
                                      value_to_set=120000)
 
1698
 
 
1699
    def test_interval_short(self):
 
1700
        self.assert_command_from_args(["-i", "PT2M", "client"],
 
1701
                                      command.SetInterval,
 
1702
                                      value_to_set=120000)
 
1703
 
 
1704
    def test_approval_delay(self):
 
1705
        self.assert_command_from_args(["--approval-delay", "PT30S",
 
1706
                                       "client"],
 
1707
                                      command.SetApprovalDelay,
 
1708
                                      value_to_set=30000)
 
1709
 
 
1710
    def test_approval_duration(self):
 
1711
        self.assert_command_from_args(["--approval-duration", "PT1S",
 
1712
                                       "client"],
 
1713
                                      command.SetApprovalDuration,
 
1714
                                      value_to_set=1000)
 
1715
 
 
1716
    def test_print_table(self):
 
1717
        self.assert_command_from_args([], command.PrintTable,
 
1718
                                      verbose=False)
 
1719
 
 
1720
    def test_print_table_verbose(self):
 
1721
        self.assert_command_from_args(["--verbose"],
 
1722
                                      command.PrintTable,
 
1723
                                      verbose=True)
 
1724
 
 
1725
    def test_print_table_verbose_short(self):
 
1726
        self.assert_command_from_args(["-v"], command.PrintTable,
 
1727
                                      verbose=True)
 
1728
 
 
1729
 
 
1730
class TestCommand(unittest.TestCase):
 
1731
    """Abstract class for tests of command classes"""
 
1732
 
 
1733
    class FakeMandosBus(dbus.MandosBus):
 
1734
        def __init__(self, testcase):
 
1735
            self.client_properties = {
 
1736
                "Name": "foo",
 
1737
                "KeyID": ("92ed150794387c03ce684574b1139a65"
 
1738
                          "94a34f895daaaf09fd8ea90a27cddb12"),
 
1739
                "Secret": b"secret",
 
1740
                "Host": "foo.example.org",
 
1741
                "Enabled": True,
 
1742
                "Timeout": 300000,
 
1743
                "LastCheckedOK": "2019-02-03T00:00:00",
 
1744
                "Created": "2019-01-02T00:00:00",
 
1745
                "Interval": 120000,
 
1746
                "Fingerprint": ("778827225BA7DE539C5A"
 
1747
                                "7CFA59CFF7CDBD9A5920"),
 
1748
                "CheckerRunning": False,
 
1749
                "LastEnabled": "2019-01-03T00:00:00",
 
1750
                "ApprovalPending": False,
 
1751
                "ApprovedByDefault": True,
 
1752
                "LastApprovalRequest": "",
 
1753
                "ApprovalDelay": 0,
 
1754
                "ApprovalDuration": 1000,
 
1755
                "Checker": "fping -q -- %(host)s",
 
1756
                "ExtendedTimeout": 900000,
 
1757
                "Expires": "2019-02-04T00:00:00",
 
1758
                "LastCheckerStatus": 0,
 
1759
            }
 
1760
            self.other_client_properties = {
 
1761
                "Name": "barbar",
 
1762
                "KeyID": ("0558568eedd67d622f5c83b35a115f79"
 
1763
                          "6ab612cff5ad227247e46c2b020f441c"),
 
1764
                "Secret": b"secretbar",
 
1765
                "Host": "192.0.2.3",
 
1766
                "Enabled": True,
 
1767
                "Timeout": 300000,
 
1768
                "LastCheckedOK": "2019-02-04T00:00:00",
 
1769
                "Created": "2019-01-03T00:00:00",
 
1770
                "Interval": 120000,
 
1771
                "Fingerprint": ("3E393AEAEFB84C7E89E2"
 
1772
                                "F547B3A107558FCA3A27"),
 
1773
                "CheckerRunning": True,
 
1774
                "LastEnabled": "2019-01-04T00:00:00",
 
1775
                "ApprovalPending": False,
 
1776
                "ApprovedByDefault": False,
 
1777
                "LastApprovalRequest": "2019-01-03T00:00:00",
 
1778
                "ApprovalDelay": 30000,
 
1779
                "ApprovalDuration": 93785000,
 
1780
                "Checker": ":",
 
1781
                "ExtendedTimeout": 900000,
 
1782
                "Expires": "2019-02-05T00:00:00",
 
1783
                "LastCheckerStatus": -2,
 
1784
            }
 
1785
            self.clients =  collections.OrderedDict(
 
1786
                [
 
1787
                    ("client_objectpath", self.client_properties),
 
1788
                    ("other_client_objectpath",
 
1789
                     self.other_client_properties),
 
1790
                ])
 
1791
            self.one_client = {"client_objectpath":
 
1792
                               self.client_properties}
 
1793
            self.testcase = testcase
 
1794
            self.calls = []
 
1795
 
 
1796
        def call_method(self, methodname, busname, objectpath,
 
1797
                        interface, *args):
 
1798
            self.testcase.assertEqual("se.recompile.Mandos", busname)
 
1799
            self.calls.append((methodname, busname, objectpath,
 
1800
                               interface, args))
 
1801
            if interface == "org.freedesktop.DBus.Properties":
 
1802
                if methodname == "Set":
 
1803
                    self.testcase.assertEqual(3, len(args))
 
1804
                    interface, key, value = args
 
1805
                    self.testcase.assertEqual(
 
1806
                        "se.recompile.Mandos.Client", interface)
 
1807
                    self.clients[objectpath][key] = value
 
1808
                    return
 
1809
            elif interface == "se.recompile.Mandos":
 
1810
                self.testcase.assertEqual("RemoveClient", methodname)
 
1811
                self.testcase.assertEqual(1, len(args))
 
1812
                clientpath = args[0]
 
1813
                del self.clients[clientpath]
 
1814
                return
 
1815
            elif interface == "se.recompile.Mandos.Client":
 
1816
                if methodname == "Approve":
 
1817
                    self.testcase.assertEqual(1, len(args))
 
1818
                    return
 
1819
            raise ValueError()
 
1820
 
 
1821
    def setUp(self):
 
1822
        self.bus = self.FakeMandosBus(self)
 
1823
 
 
1824
 
 
1825
class TestBaseCommands(TestCommand):
 
1826
 
 
1827
    def test_IsEnabled_exits_successfully(self):
 
1828
        with self.assertRaises(SystemExit) as e:
 
1829
            command.IsEnabled().run(self.bus.one_client)
 
1830
        if e.exception.code is not None:
 
1831
            self.assertEqual(0, e.exception.code)
 
1832
        else:
 
1833
            self.assertIsNone(e.exception.code)
 
1834
 
 
1835
    def test_IsEnabled_exits_with_failure(self):
 
1836
        self.bus.client_properties["Enabled"] = False
 
1837
        with self.assertRaises(SystemExit) as e:
 
1838
            command.IsEnabled().run(self.bus.one_client)
 
1839
        if isinstance(e.exception.code, int):
 
1840
            self.assertNotEqual(0, e.exception.code)
 
1841
        else:
 
1842
            self.assertIsNotNone(e.exception.code)
 
1843
 
 
1844
    def test_Approve(self):
 
1845
        busname = "se.recompile.Mandos"
 
1846
        client_interface = "se.recompile.Mandos.Client"
 
1847
        command.Approve().run(self.bus.clients, self.bus)
 
1848
        for clientpath in self.bus.clients:
 
1849
            self.assertIn(("Approve", busname, clientpath,
 
1850
                           client_interface, (True,)), self.bus.calls)
 
1851
 
 
1852
    def test_Deny(self):
 
1853
        busname = "se.recompile.Mandos"
 
1854
        client_interface = "se.recompile.Mandos.Client"
 
1855
        command.Deny().run(self.bus.clients, self.bus)
 
1856
        for clientpath in self.bus.clients:
 
1857
            self.assertIn(("Approve", busname, clientpath,
 
1858
                           client_interface, (False,)),
 
1859
                          self.bus.calls)
 
1860
 
 
1861
    def test_Remove(self):
 
1862
        command.Remove().run(self.bus.clients, self.bus)
 
1863
        for clientpath in self.bus.clients:
 
1864
            self.assertIn(("RemoveClient", dbus_busname,
 
1865
                           dbus_server_path, dbus_server_interface,
 
1866
                           (clientpath,)), self.bus.calls)
 
1867
 
 
1868
    expected_json = {
 
1869
        "foo": {
 
1870
            "Name": "foo",
 
1871
            "KeyID": ("92ed150794387c03ce684574b1139a65"
 
1872
                      "94a34f895daaaf09fd8ea90a27cddb12"),
 
1873
            "Host": "foo.example.org",
 
1874
            "Enabled": True,
 
1875
            "Timeout": 300000,
 
1876
            "LastCheckedOK": "2019-02-03T00:00:00",
 
1877
            "Created": "2019-01-02T00:00:00",
 
1878
            "Interval": 120000,
 
1879
            "Fingerprint": ("778827225BA7DE539C5A"
 
1880
                            "7CFA59CFF7CDBD9A5920"),
 
1881
            "CheckerRunning": False,
 
1882
            "LastEnabled": "2019-01-03T00:00:00",
 
1883
            "ApprovalPending": False,
 
1884
            "ApprovedByDefault": True,
 
1885
            "LastApprovalRequest": "",
 
1886
            "ApprovalDelay": 0,
 
1887
            "ApprovalDuration": 1000,
 
1888
            "Checker": "fping -q -- %(host)s",
 
1889
            "ExtendedTimeout": 900000,
 
1890
            "Expires": "2019-02-04T00:00:00",
 
1891
            "LastCheckerStatus": 0,
 
1892
        },
 
1893
        "barbar": {
 
1894
            "Name": "barbar",
 
1895
            "KeyID": ("0558568eedd67d622f5c83b35a115f79"
 
1896
                      "6ab612cff5ad227247e46c2b020f441c"),
 
1897
            "Host": "192.0.2.3",
 
1898
            "Enabled": True,
 
1899
            "Timeout": 300000,
 
1900
            "LastCheckedOK": "2019-02-04T00:00:00",
 
1901
            "Created": "2019-01-03T00:00:00",
 
1902
            "Interval": 120000,
 
1903
            "Fingerprint": ("3E393AEAEFB84C7E89E2"
 
1904
                            "F547B3A107558FCA3A27"),
 
1905
            "CheckerRunning": True,
 
1906
            "LastEnabled": "2019-01-04T00:00:00",
 
1907
            "ApprovalPending": False,
 
1908
            "ApprovedByDefault": False,
 
1909
            "LastApprovalRequest": "2019-01-03T00:00:00",
 
1910
            "ApprovalDelay": 30000,
 
1911
            "ApprovalDuration": 93785000,
 
1912
            "Checker": ":",
 
1913
            "ExtendedTimeout": 900000,
 
1914
            "Expires": "2019-02-05T00:00:00",
 
1915
            "LastCheckerStatus": -2,
 
1916
        },
 
1917
    }
 
1918
 
 
1919
    def test_DumpJSON_normal(self):
 
1920
        with self.capture_stdout_to_buffer() as buffer:
 
1921
            command.DumpJSON().run(self.bus.clients)
 
1922
        json_data = json.loads(buffer.getvalue())
 
1923
        self.assertDictEqual(self.expected_json, json_data)
 
1924
 
 
1925
    @staticmethod
 
1926
    @contextlib.contextmanager
 
1927
    def capture_stdout_to_buffer():
 
1928
        capture_buffer = io.StringIO()
 
1929
        old_stdout = sys.stdout
 
1930
        sys.stdout = capture_buffer
 
1931
        try:
 
1932
            yield capture_buffer
 
1933
        finally:
 
1934
            sys.stdout = old_stdout
 
1935
 
 
1936
    def test_DumpJSON_one_client(self):
 
1937
        with self.capture_stdout_to_buffer() as buffer:
 
1938
            command.DumpJSON().run(self.bus.one_client)
 
1939
        json_data = json.loads(buffer.getvalue())
 
1940
        expected_json = {"foo": self.expected_json["foo"]}
 
1941
        self.assertDictEqual(expected_json, json_data)
 
1942
 
 
1943
    def test_PrintTable_normal(self):
 
1944
        with self.capture_stdout_to_buffer() as buffer:
 
1945
            command.PrintTable().run(self.bus.clients)
 
1946
        expected_output = "\n".join((
 
1947
            "Name   Enabled Timeout  Last Successful Check",
 
1948
            "foo    Yes     00:05:00 2019-02-03T00:00:00  ",
 
1949
            "barbar Yes     00:05:00 2019-02-04T00:00:00  ",
 
1950
        )) + "\n"
 
1951
        self.assertEqual(expected_output, buffer.getvalue())
 
1952
 
 
1953
    def test_PrintTable_verbose(self):
 
1954
        with self.capture_stdout_to_buffer() as buffer:
 
1955
            command.PrintTable(verbose=True).run(self.bus.clients)
 
1956
        columns = (
 
1957
            (
 
1958
                "Name   ",
 
1959
                "foo    ",
 
1960
                "barbar ",
 
1961
            ),(
 
1962
                "Enabled ",
 
1963
                "Yes     ",
 
1964
                "Yes     ",
 
1965
            ),(
 
1966
                "Timeout  ",
 
1967
                "00:05:00 ",
 
1968
                "00:05:00 ",
 
1969
            ),(
 
1970
                "Last Successful Check ",
 
1971
                "2019-02-03T00:00:00   ",
 
1972
                "2019-02-04T00:00:00   ",
 
1973
            ),(
 
1974
                "Created             ",
 
1975
                "2019-01-02T00:00:00 ",
 
1976
                "2019-01-03T00:00:00 ",
 
1977
            ),(
 
1978
                "Interval ",
 
1979
                "00:02:00 ",
 
1980
                "00:02:00 ",
 
1981
            ),(
 
1982
                "Host            ",
 
1983
                "foo.example.org ",
 
1984
                "192.0.2.3       ",
 
1985
            ),(
 
1986
                ("Key ID                                             "
 
1987
                 "              "),
 
1988
                ("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
 
1989
                 "ea90a27cddb12 "),
 
1990
                ("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
 
1991
                 "46c2b020f441c "),
 
1992
            ),(
 
1993
                "Fingerprint                              ",
 
1994
                "778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
 
1995
                "3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
 
1996
            ),(
 
1997
                "Check Is Running ",
 
1998
                "No               ",
 
1999
                "Yes              ",
 
2000
            ),(
 
2001
                "Last Enabled        ",
 
2002
                "2019-01-03T00:00:00 ",
 
2003
                "2019-01-04T00:00:00 ",
 
2004
            ),(
 
2005
                "Approval Is Pending ",
 
2006
                "No                  ",
 
2007
                "No                  ",
 
2008
            ),(
 
2009
                "Approved By Default ",
 
2010
                "Yes                 ",
 
2011
                "No                  ",
 
2012
            ),(
 
2013
                "Last Approval Request ",
 
2014
                "                      ",
 
2015
                "2019-01-03T00:00:00   ",
 
2016
            ),(
 
2017
                "Approval Delay ",
 
2018
                "00:00:00       ",
 
2019
                "00:00:30       ",
 
2020
            ),(
 
2021
                "Approval Duration ",
 
2022
                "00:00:01          ",
 
2023
                "1T02:03:05        ",
 
2024
            ),(
 
2025
                "Checker              ",
 
2026
                "fping -q -- %(host)s ",
 
2027
                ":                    ",
 
2028
            ),(
 
2029
                "Extended Timeout ",
 
2030
                "00:15:00         ",
 
2031
                "00:15:00         ",
 
2032
            ),(
 
2033
                "Expires             ",
 
2034
                "2019-02-04T00:00:00 ",
 
2035
                "2019-02-05T00:00:00 ",
 
2036
            ),(
 
2037
                "Last Checker Status",
 
2038
                "0                  ",
 
2039
                "-2                 ",
 
2040
            )
 
2041
        )
 
2042
        num_lines = max(len(rows) for rows in columns)
 
2043
        expected_output = ("\n".join("".join(rows[line]
 
2044
                                             for rows in columns)
 
2045
                                     for line in range(num_lines))
 
2046
                           + "\n")
 
2047
        self.assertEqual(expected_output, buffer.getvalue())
 
2048
 
 
2049
    def test_PrintTable_one_client(self):
 
2050
        with self.capture_stdout_to_buffer() as buffer:
 
2051
            command.PrintTable().run(self.bus.one_client)
 
2052
        expected_output = "\n".join((
 
2053
            "Name Enabled Timeout  Last Successful Check",
 
2054
            "foo  Yes     00:05:00 2019-02-03T00:00:00  ",
 
2055
        )) + "\n"
 
2056
        self.assertEqual(expected_output, buffer.getvalue())
 
2057
 
 
2058
 
 
2059
class TestPropertySetterCmd(TestCommand):
 
2060
    """Abstract class for tests of command.PropertySetter classes"""
 
2061
 
 
2062
    def runTest(self):
 
2063
        if not hasattr(self, "command"):
 
2064
            return              # Abstract TestCase class
 
2065
 
 
2066
        if hasattr(self, "values_to_set"):
 
2067
            cmd_args = [(value,) for value in self.values_to_set]
 
2068
            values_to_get = getattr(self, "values_to_get",
 
2069
                                    self.values_to_set)
 
2070
        else:
 
2071
            cmd_args = [() for x in range(len(self.values_to_get))]
 
2072
            values_to_get = self.values_to_get
 
2073
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
 
2074
            for clientpath in self.bus.clients:
 
2075
                self.bus.clients[clientpath][self.propname] = (
 
2076
                    Unique())
 
2077
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
 
2078
            for clientpath in self.bus.clients:
 
2079
                value = (self.bus.clients[clientpath]
 
2080
                         [self.propname])
 
2081
                self.assertNotIsInstance(value, Unique)
 
2082
                self.assertEqual(value_to_get, value)
 
2083
 
 
2084
 
 
2085
class TestEnableCmd(TestPropertySetterCmd):
 
2086
    command = command.Enable
 
2087
    propname = "Enabled"
 
2088
    values_to_get = [True]
 
2089
 
 
2090
 
 
2091
class TestDisableCmd(TestPropertySetterCmd):
 
2092
    command = command.Disable
 
2093
    propname = "Enabled"
 
2094
    values_to_get = [False]
 
2095
 
 
2096
 
 
2097
class TestBumpTimeoutCmd(TestPropertySetterCmd):
 
2098
    command = command.BumpTimeout
 
2099
    propname = "LastCheckedOK"
 
2100
    values_to_get = [""]
 
2101
 
 
2102
 
 
2103
class TestStartCheckerCmd(TestPropertySetterCmd):
 
2104
    command = command.StartChecker
 
2105
    propname = "CheckerRunning"
 
2106
    values_to_get = [True]
 
2107
 
 
2108
 
 
2109
class TestStopCheckerCmd(TestPropertySetterCmd):
 
2110
    command = command.StopChecker
 
2111
    propname = "CheckerRunning"
 
2112
    values_to_get = [False]
 
2113
 
 
2114
 
 
2115
class TestApproveByDefaultCmd(TestPropertySetterCmd):
 
2116
    command = command.ApproveByDefault
 
2117
    propname = "ApprovedByDefault"
 
2118
    values_to_get = [True]
 
2119
 
 
2120
 
 
2121
class TestDenyByDefaultCmd(TestPropertySetterCmd):
 
2122
    command = command.DenyByDefault
 
2123
    propname = "ApprovedByDefault"
 
2124
    values_to_get = [False]
 
2125
 
 
2126
 
 
2127
class TestSetCheckerCmd(TestPropertySetterCmd):
 
2128
    command = command.SetChecker
 
2129
    propname = "Checker"
 
2130
    values_to_set = ["", ":", "fping -q -- %s"]
 
2131
 
 
2132
 
 
2133
class TestSetHostCmd(TestPropertySetterCmd):
 
2134
    command = command.SetHost
 
2135
    propname = "Host"
 
2136
    values_to_set = ["192.0.2.3", "client.example.org"]
 
2137
 
 
2138
 
 
2139
class TestSetSecretCmd(TestPropertySetterCmd):
 
2140
    command = command.SetSecret
 
2141
    propname = "Secret"
 
2142
    values_to_set = [io.BytesIO(b""),
 
2143
                     io.BytesIO(b"secret\0xyzzy\nbar")]
 
2144
    values_to_get = [f.getvalue() for f in values_to_set]
 
2145
 
 
2146
 
 
2147
class TestSetTimeoutCmd(TestPropertySetterCmd):
 
2148
    command = command.SetTimeout
 
2149
    propname = "Timeout"
 
2150
    values_to_set = [datetime.timedelta(),
 
2151
                     datetime.timedelta(minutes=5),
 
2152
                     datetime.timedelta(seconds=1),
 
2153
                     datetime.timedelta(weeks=1),
 
2154
                     datetime.timedelta(weeks=52)]
 
2155
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2156
 
 
2157
 
 
2158
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
 
2159
    command = command.SetExtendedTimeout
 
2160
    propname = "ExtendedTimeout"
 
2161
    values_to_set = [datetime.timedelta(),
 
2162
                     datetime.timedelta(minutes=5),
 
2163
                     datetime.timedelta(seconds=1),
 
2164
                     datetime.timedelta(weeks=1),
 
2165
                     datetime.timedelta(weeks=52)]
 
2166
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2167
 
 
2168
 
 
2169
class TestSetIntervalCmd(TestPropertySetterCmd):
 
2170
    command = command.SetInterval
 
2171
    propname = "Interval"
 
2172
    values_to_set = [datetime.timedelta(),
 
2173
                     datetime.timedelta(minutes=5),
 
2174
                     datetime.timedelta(seconds=1),
 
2175
                     datetime.timedelta(weeks=1),
 
2176
                     datetime.timedelta(weeks=52)]
 
2177
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2178
 
 
2179
 
 
2180
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
 
2181
    command = command.SetApprovalDelay
 
2182
    propname = "ApprovalDelay"
 
2183
    values_to_set = [datetime.timedelta(),
 
2184
                     datetime.timedelta(minutes=5),
 
2185
                     datetime.timedelta(seconds=1),
 
2186
                     datetime.timedelta(weeks=1),
 
2187
                     datetime.timedelta(weeks=52)]
 
2188
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2189
 
 
2190
 
 
2191
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
 
2192
    command = command.SetApprovalDuration
 
2193
    propname = "ApprovalDuration"
 
2194
    values_to_set = [datetime.timedelta(),
 
2195
                     datetime.timedelta(minutes=5),
 
2196
                     datetime.timedelta(seconds=1),
 
2197
                     datetime.timedelta(weeks=1),
 
2198
                     datetime.timedelta(weeks=52)]
 
2199
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2200
 
 
2201
 
 
2202
 
 
2203
def should_only_run_tests():
 
2204
    parser = argparse.ArgumentParser(add_help=False)
 
2205
    parser.add_argument("--check", action='store_true')
 
2206
    args, unknown_args = parser.parse_known_args()
 
2207
    run_tests = args.check
 
2208
    if run_tests:
 
2209
        # Remove --check argument from sys.argv
 
2210
        sys.argv[1:] = unknown_args
 
2211
    return run_tests
 
2212
 
 
2213
# Add all tests from doctest strings
 
2214
def load_tests(loader, tests, none):
 
2215
    import doctest
 
2216
    tests.addTests(doctest.DocTestSuite())
 
2217
    return tests
 
2218
 
 
2219
if __name__ == "__main__":
 
2220
    try:
 
2221
        if should_only_run_tests():
 
2222
            # Call using ./tdd-python-script --check [--verbose]
 
2223
            unittest.main()
 
2224
        else:
 
2225
            main()
 
2226
    finally:
 
2227
        logging.shutdown()