/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-15 23:55:53 UTC
  • mto: This revision was merged to the branch mainline in revision 382.
  • Revision ID: teddy@recompile.se-20190315235553-znbpfn7d8o84tyt6
mandos-ctl: Refactor

* mandos-ctl (main): Do some minor refactoring.

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