/mandos/trunk

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

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-17 17:34:07 UTC
  • Revision ID: teddy@recompile.se-20190317173407-itlj1a7mzfwb3yz3
mandos-ctl: Refactor tests

* mandos-ctl: Change all calls to assertEqual (and related functions
              assertNotEqual and assertDictEqual) to always pass
              arguments in (expected, actual) order, for consistency.

Show diffs side-by-side

added added

removed removed

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