/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: 2024-09-08 03:27:08 UTC
  • Revision ID: teddy@recompile.se-20240908032708-d0mwgln797i3ddat
Use new package name "libgnutls30t64" when available

* debian/control (Package: mandos/Depends): Add "libgnutls30t64" as an
  alternative to libgnutls30.

Reported-By: Bastian Germann <Bastian.Germann@gmx.de>
Suggested-by: Bastian Germann <Bastian.Germann@gmx.de>

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
#!/usr/bin/python
2
 
# -*- mode: python; coding: utf-8 -*-
3
 
4
 
# Mandos Monitor - Control and monitor the Mandos server
5
 
6
 
# Copyright © 2008-2011 Teddy Hogeborn
7
 
# Copyright © 2008-2011 Björn Påhlsson
8
 
9
 
# This program is free software: you can redistribute it and/or modify
10
 
# it under the terms of the GNU General Public License as published by
 
1
#!/usr/bin/python3 -bbI
 
2
# -*- coding: utf-8; lexical-binding: t -*-
 
3
#
 
4
# Mandos Control - Control or query the Mandos server
 
5
#
 
6
# Copyright © 2008-2022 Teddy Hogeborn
 
7
# Copyright © 2008-2022 Björn Påhlsson
 
8
#
 
9
# This file is part of Mandos.
 
10
#
 
11
# Mandos is free software: you can redistribute it and/or modify it
 
12
# under the terms of the GNU General Public License as published by
11
13
# the Free Software Foundation, either version 3 of the License, or
12
14
# (at your option) any later version.
13
15
#
14
 
#     This program is distributed in the hope that it will be useful,
15
 
#     but WITHOUT ANY WARRANTY; without even the implied warranty of
 
16
#     Mandos is distributed in the hope that it will be useful, but
 
17
#     WITHOUT ANY WARRANTY; without even the implied warranty of
16
18
#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17
19
#     GNU General Public License for more details.
18
 
 
20
#
19
21
# You should have received a copy of the GNU General Public License
20
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
21
 
 
22
# along with Mandos.  If not, see <http://www.gnu.org/licenses/>.
 
23
#
22
24
# Contact the authors at <mandos@recompile.se>.
23
 
24
 
 
 
25
#
25
26
from __future__ import (division, absolute_import, print_function,
26
27
                        unicode_literals)
27
28
 
 
29
try:
 
30
    from future_builtins import *
 
31
except ImportError:
 
32
    pass
 
33
 
28
34
import sys
29
 
import dbus
 
35
import unittest
30
36
import argparse
 
37
import logging
 
38
import os
31
39
import locale
32
40
import datetime
33
41
import re
34
 
import os
 
42
import collections
 
43
import json
 
44
import io
 
45
import tempfile
 
46
import contextlib
 
47
 
 
48
if sys.version_info.major == 2:
 
49
    __metaclass__ = type
 
50
    str = unicode
 
51
    input = raw_input
 
52
 
 
53
 
 
54
class gi:
 
55
    """Dummy gi module, for the tests"""
 
56
    class repository:
 
57
        class GLib:
 
58
            class Error(Exception):
 
59
                pass
 
60
 
 
61
 
 
62
dbussy = None
 
63
ravel = None
 
64
dbus_python = None
 
65
pydbus = None
 
66
 
 
67
try:
 
68
    import dbussy
 
69
    import ravel
 
70
except ImportError:
 
71
    try:
 
72
        import pydbus
 
73
        import gi
 
74
    except ImportError:
 
75
        import dbus as dbus_python
 
76
 
 
77
 
 
78
# Show warnings by default
 
79
if not sys.warnoptions:
 
80
    import warnings
 
81
    warnings.simplefilter("default")
 
82
 
 
83
log = logging.getLogger(os.path.basename(sys.argv[0]))
 
84
logging.basicConfig(level="INFO",         # Show info level messages
 
85
                    format="%(message)s") # Show basic log messages
 
86
 
 
87
logging.captureWarnings(True)   # Show warnings via the logging system
 
88
 
 
89
if sys.version_info.major == 2:
 
90
    import StringIO
 
91
    io.StringIO = StringIO.StringIO
35
92
 
36
93
locale.setlocale(locale.LC_ALL, "")
37
94
 
38
 
tablewords = {
39
 
    "Name": "Name",
40
 
    "Enabled": "Enabled",
41
 
    "Timeout": "Timeout",
42
 
    "LastCheckedOK": "Last Successful Check",
43
 
    "LastApprovalRequest": "Last Approval Request",
44
 
    "Created": "Created",
45
 
    "Interval": "Interval",
46
 
    "Host": "Host",
47
 
    "Fingerprint": "Fingerprint",
48
 
    "CheckerRunning": "Check Is Running",
49
 
    "LastEnabled": "Last Enabled",
50
 
    "ApprovalPending": "Approval Is Pending",
51
 
    "ApprovedByDefault": "Approved By Default",
52
 
    "ApprovalDelay": "Approval Delay",
53
 
    "ApprovalDuration": "Approval Duration",
54
 
    "Checker": "Checker",
55
 
    "ExtendedTimeout" : "Extended Timeout"
56
 
    }
57
 
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
58
 
domain = "se.recompile"
59
 
busname = domain + ".Mandos"
60
 
server_path = "/"
61
 
server_interface = domain + ".Mandos"
62
 
client_interface = domain + ".Mandos.Client"
63
 
version = "1.4.1"
64
 
 
65
 
def timedelta_to_milliseconds(td):
66
 
    """Convert a datetime.timedelta object to milliseconds"""
67
 
    return ((td.days * 24 * 60 * 60 * 1000)
68
 
            + (td.seconds * 1000)
69
 
            + (td.microseconds // 1000))
70
 
 
71
 
def milliseconds_to_string(ms):
72
 
    td = datetime.timedelta(0, 0, 0, ms)
73
 
    return ("%(days)s%(hours)02d:%(minutes)02d:%(seconds)02d"
74
 
            % { "days": "%dT" % td.days if td.days else "",
75
 
                "hours": td.seconds // 3600,
76
 
                "minutes": (td.seconds % 3600) // 60,
77
 
                "seconds": td.seconds % 60,
78
 
                })
79
 
 
80
 
def string_to_delta(interval):
81
 
    """Parse a string and return a datetime.timedelta
82
 
    
83
 
    >>> string_to_delta("7d")
84
 
    datetime.timedelta(7)
85
 
    >>> string_to_delta("60s")
86
 
    datetime.timedelta(0, 60)
87
 
    >>> string_to_delta("60m")
88
 
    datetime.timedelta(0, 3600)
89
 
    >>> string_to_delta("24h")
90
 
    datetime.timedelta(1)
91
 
    >>> string_to_delta("1w")
92
 
    datetime.timedelta(7)
93
 
    >>> string_to_delta("5m 30s")
94
 
    datetime.timedelta(0, 330)
95
 
    """
96
 
    timevalue = datetime.timedelta(0)
97
 
    regexp = re.compile("\d+[dsmhw]")
98
 
    
99
 
    for s in regexp.findall(interval):
100
 
        try:
101
 
            suffix = unicode(s[-1])
102
 
            value = int(s[:-1])
103
 
            if suffix == "d":
104
 
                delta = datetime.timedelta(value)
105
 
            elif suffix == "s":
106
 
                delta = datetime.timedelta(0, value)
107
 
            elif suffix == "m":
108
 
                delta = datetime.timedelta(0, 0, 0, 0, value)
109
 
            elif suffix == "h":
110
 
                delta = datetime.timedelta(0, 0, 0, 0, 0, value)
111
 
            elif suffix == "w":
112
 
                delta = datetime.timedelta(0, 0, 0, 0, 0, 0, value)
113
 
            else:
114
 
                raise ValueError
115
 
        except (ValueError, IndexError):
116
 
            raise ValueError
117
 
        timevalue += delta
118
 
    return timevalue
119
 
 
120
 
def print_clients(clients, keywords):
121
 
    def valuetostring(value, keyword):
122
 
        if type(value) is dbus.Boolean:
123
 
            return "Yes" if value else "No"
124
 
        if keyword in ("Timeout", "Interval", "ApprovalDelay",
125
 
                       "ApprovalDuration"):
126
 
            return milliseconds_to_string(value)
127
 
        return unicode(value)
128
 
    
129
 
    # Create format string to print table rows
130
 
    format_string = " ".join("%%-%ds" %
131
 
                             max(len(tablewords[key]),
132
 
                                 max(len(valuetostring(client[key],
133
 
                                                       key))
134
 
                                     for client in
135
 
                                     clients))
136
 
                             for key in keywords)
137
 
    # Print header line
138
 
    print(format_string % tuple(tablewords[key] for key in keywords))
139
 
    for client in clients:
140
 
        print(format_string % tuple(valuetostring(client[key], key)
141
 
                                    for key in keywords))
142
 
 
143
 
def has_actions(options):
144
 
    return any((options.enable,
145
 
                options.disable,
146
 
                options.bump_timeout,
147
 
                options.start_checker,
148
 
                options.stop_checker,
149
 
                options.is_enabled,
150
 
                options.remove,
151
 
                options.checker is not None,
152
 
                options.timeout is not None,
153
 
                options.extended_timeout is not None,
154
 
                options.interval is not None,
155
 
                options.approved_by_default is not None,
156
 
                options.approval_delay is not None,
157
 
                options.approval_duration is not None,
158
 
                options.host is not None,
159
 
                options.secret is not None,
160
 
                options.approve,
161
 
                options.deny))
 
95
version = "1.8.16"
 
96
 
162
97
 
163
98
def main():
164
99
    parser = argparse.ArgumentParser()
 
100
    add_command_line_options(parser)
 
101
 
 
102
    options = parser.parse_args()
 
103
    check_option_syntax(parser, options)
 
104
 
 
105
    clientnames = options.client
 
106
 
 
107
    if options.debug:
 
108
        logging.getLogger("").setLevel(logging.DEBUG)
 
109
 
 
110
    if dbussy is not None and ravel is not None:
 
111
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
 
112
    elif pydbus is not None:
 
113
        bus = pydbus_adapter.CachingBus(pydbus)
 
114
    else:
 
115
        bus = dbus_python_adapter.CachingBus(dbus_python)
 
116
 
 
117
    try:
 
118
        all_clients = bus.get_clients_and_properties()
 
119
    except dbus.ConnectFailed as e:
 
120
        log.critical("Could not connect to Mandos server: %s", e)
 
121
        sys.exit(1)
 
122
    except dbus.Error as e:
 
123
        log.critical(
 
124
            "Failed to access Mandos server through D-Bus:\n%s", e)
 
125
        sys.exit(1)
 
126
 
 
127
    # Compile dict of (clientpath: properties) to process
 
128
    if not clientnames:
 
129
        clients = all_clients
 
130
    else:
 
131
        clients = {}
 
132
        for name in clientnames:
 
133
            for objpath, properties in all_clients.items():
 
134
                if properties["Name"] == name:
 
135
                    clients[objpath] = properties
 
136
                    break
 
137
            else:
 
138
                log.critical("Client not found on server: %r", name)
 
139
                sys.exit(1)
 
140
 
 
141
    commands = commands_from_options(options)
 
142
 
 
143
    for command in commands:
 
144
        command.run(clients, bus)
 
145
 
 
146
 
 
147
def add_command_line_options(parser):
165
148
    parser.add_argument("--version", action="version",
166
 
                        version = "%%prog %s" % version,
 
149
                        version="%(prog)s {}".format(version),
167
150
                        help="show version number and exit")
168
151
    parser.add_argument("-a", "--all", action="store_true",
169
152
                        help="Select all clients")
170
153
    parser.add_argument("-v", "--verbose", action="store_true",
171
154
                        help="Print all fields")
172
 
    parser.add_argument("-e", "--enable", action="store_true",
173
 
                        help="Enable client")
174
 
    parser.add_argument("-d", "--disable", action="store_true",
175
 
                        help="disable client")
176
 
    parser.add_argument("-b", "--bump-timeout", action="store_true",
 
155
    parser.add_argument("-j", "--dump-json", dest="commands",
 
156
                        action="append_const", default=[],
 
157
                        const=command.DumpJSON(),
 
158
                        help="Dump client data in JSON format")
 
159
    enable_disable = parser.add_mutually_exclusive_group()
 
160
    enable_disable.add_argument("-e", "--enable", dest="commands",
 
161
                                action="append_const", default=[],
 
162
                                const=command.Enable(),
 
163
                                help="Enable client")
 
164
    enable_disable.add_argument("-d", "--disable", dest="commands",
 
165
                                action="append_const", default=[],
 
166
                                const=command.Disable(),
 
167
                                help="disable client")
 
168
    parser.add_argument("-b", "--bump-timeout", dest="commands",
 
169
                        action="append_const", default=[],
 
170
                        const=command.BumpTimeout(),
177
171
                        help="Bump timeout for client")
178
 
    parser.add_argument("--start-checker", action="store_true",
179
 
                        help="Start checker for client")
180
 
    parser.add_argument("--stop-checker", action="store_true",
181
 
                        help="Stop checker for client")
182
 
    parser.add_argument("-V", "--is-enabled", action="store_true",
 
172
    start_stop_checker = parser.add_mutually_exclusive_group()
 
173
    start_stop_checker.add_argument("--start-checker",
 
174
                                    dest="commands",
 
175
                                    action="append_const", default=[],
 
176
                                    const=command.StartChecker(),
 
177
                                    help="Start checker for client")
 
178
    start_stop_checker.add_argument("--stop-checker", dest="commands",
 
179
                                    action="append_const", default=[],
 
180
                                    const=command.StopChecker(),
 
181
                                    help="Stop checker for client")
 
182
    parser.add_argument("-V", "--is-enabled", dest="commands",
 
183
                        action="append_const", default=[],
 
184
                        const=command.IsEnabled(),
183
185
                        help="Check if client is enabled")
184
 
    parser.add_argument("-r", "--remove", action="store_true",
 
186
    parser.add_argument("-r", "--remove", dest="commands",
 
187
                        action="append_const", default=[],
 
188
                        const=command.Remove(),
185
189
                        help="Remove client")
186
 
    parser.add_argument("-c", "--checker",
 
190
    parser.add_argument("-c", "--checker", dest="commands",
 
191
                        action="append", default=[],
 
192
                        metavar="COMMAND", type=command.SetChecker,
187
193
                        help="Set checker command for client")
188
 
    parser.add_argument("-t", "--timeout",
189
 
                        help="Set timeout for client")
190
 
    parser.add_argument("--extended-timeout",
191
 
                        help="Set extended timeout for client")
192
 
    parser.add_argument("-i", "--interval",
193
 
                        help="Set checker interval for client")
194
 
    parser.add_argument("--approve-by-default", action="store_true",
195
 
                        default=None, dest="approved_by_default",
196
 
                        help="Set client to be approved by default")
197
 
    parser.add_argument("--deny-by-default", action="store_false",
198
 
                        dest="approved_by_default",
199
 
                        help="Set client to be denied by default")
200
 
    parser.add_argument("--approval-delay",
201
 
                        help="Set delay before client approve/deny")
202
 
    parser.add_argument("--approval-duration",
203
 
                        help="Set duration of one client approval")
204
 
    parser.add_argument("-H", "--host", help="Set host for client")
205
 
    parser.add_argument("-s", "--secret", type=file,
206
 
                        help="Set password blob (file) for client")
207
 
    parser.add_argument("-A", "--approve", action="store_true",
208
 
                        help="Approve any current client request")
209
 
    parser.add_argument("-D", "--deny", action="store_true",
210
 
                        help="Deny any current client request")
 
194
    parser.add_argument(
 
195
        "-t", "--timeout", dest="commands", action="append",
 
196
        default=[], metavar="TIME",
 
197
        type=command.SetTimeout.argparse(string_to_delta),
 
198
        help="Set timeout for client")
 
199
    parser.add_argument(
 
200
        "--extended-timeout", dest="commands", action="append",
 
201
        default=[], metavar="TIME",
 
202
        type=command.SetExtendedTimeout.argparse(string_to_delta),
 
203
        help="Set extended timeout for client")
 
204
    parser.add_argument(
 
205
        "-i", "--interval", dest="commands", action="append",
 
206
        default=[], metavar="TIME",
 
207
        type=command.SetInterval.argparse(string_to_delta),
 
208
        help="Set checker interval for client")
 
209
    approve_deny_default = parser.add_mutually_exclusive_group()
 
210
    approve_deny_default.add_argument(
 
211
        "--approve-by-default", dest="commands",
 
212
        action="append_const", default=[],
 
213
        const=command.ApproveByDefault(),
 
214
        help="Set client to be approved by default")
 
215
    approve_deny_default.add_argument(
 
216
        "--deny-by-default", dest="commands",
 
217
        action="append_const", default=[],
 
218
        const=command.DenyByDefault(),
 
219
        help="Set client to be denied by default")
 
220
    parser.add_argument(
 
221
        "--approval-delay", dest="commands", action="append",
 
222
        default=[], metavar="TIME",
 
223
        type=command.SetApprovalDelay.argparse(string_to_delta),
 
224
        help="Set delay before client approve/deny")
 
225
    parser.add_argument(
 
226
        "--approval-duration", dest="commands", action="append",
 
227
        default=[], metavar="TIME",
 
228
        type=command.SetApprovalDuration.argparse(string_to_delta),
 
229
        help="Set duration of one client approval")
 
230
    parser.add_argument("-H", "--host", dest="commands",
 
231
                        action="append", default=[], metavar="STRING",
 
232
                        type=command.SetHost,
 
233
                        help="Set host for client")
 
234
    parser.add_argument(
 
235
        "-s", "--secret", dest="commands", action="append",
 
236
        default=[], metavar="FILENAME",
 
237
        type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
 
238
        help="Set password blob (file) for client")
 
239
    approve_deny = parser.add_mutually_exclusive_group()
 
240
    approve_deny.add_argument(
 
241
        "-A", "--approve", dest="commands", action="append_const",
 
242
        default=[], const=command.Approve(),
 
243
        help="Approve any current client request")
 
244
    approve_deny.add_argument("-D", "--deny", dest="commands",
 
245
                              action="append_const", default=[],
 
246
                              const=command.Deny(),
 
247
                              help="Deny any current client request")
 
248
    parser.add_argument("--debug", action="store_true",
 
249
                        help="Debug mode (show D-Bus commands)")
 
250
    parser.add_argument("--check", action="store_true",
 
251
                        help="Run self-test")
211
252
    parser.add_argument("client", nargs="*", help="Client name")
212
 
    options = parser.parse_args()
213
 
    
214
 
    if has_actions(options) and not options.client and not options.all:
 
253
 
 
254
 
 
255
def string_to_delta(interval):
 
256
    """Parse a string and return a datetime.timedelta"""
 
257
 
 
258
    try:
 
259
        return rfc3339_duration_to_delta(interval)
 
260
    except ValueError as e:
 
261
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
 
262
                    " ".join(e.args))
 
263
    return parse_pre_1_6_1_interval(interval)
 
264
 
 
265
 
 
266
def rfc3339_duration_to_delta(duration):
 
267
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
 
268
 
 
269
    >>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
 
270
    True
 
271
    >>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
 
272
    True
 
273
    >>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
 
274
    True
 
275
    >>> # 60 months
 
276
    >>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
 
277
    True
 
278
    >>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
 
279
    True
 
280
    >>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
 
281
    True
 
282
    >>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
 
283
    True
 
284
    >>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
 
285
    True
 
286
    >>> # Can not be empty:
 
287
    >>> rfc3339_duration_to_delta("")
 
288
    Traceback (most recent call last):
 
289
    ...
 
290
    ValueError: Invalid RFC 3339 duration: ""
 
291
    >>> # Must start with "P":
 
292
    >>> rfc3339_duration_to_delta("1D")
 
293
    Traceback (most recent call last):
 
294
    ...
 
295
    ValueError: Invalid RFC 3339 duration: "1D"
 
296
    >>> # Must use correct order
 
297
    >>> rfc3339_duration_to_delta("PT1S2M")
 
298
    Traceback (most recent call last):
 
299
    ...
 
300
    ValueError: Invalid RFC 3339 duration: "PT1S2M"
 
301
    >>> # Time needs time marker
 
302
    >>> rfc3339_duration_to_delta("P1H2S")
 
303
    Traceback (most recent call last):
 
304
    ...
 
305
    ValueError: Invalid RFC 3339 duration: "P1H2S"
 
306
    >>> # Weeks can not be combined with anything else
 
307
    >>> rfc3339_duration_to_delta("P1D2W")
 
308
    Traceback (most recent call last):
 
309
    ...
 
310
    ValueError: Invalid RFC 3339 duration: "P1D2W"
 
311
    >>> rfc3339_duration_to_delta("P2W2H")
 
312
    Traceback (most recent call last):
 
313
    ...
 
314
    ValueError: Invalid RFC 3339 duration: "P2W2H"
 
315
    """
 
316
 
 
317
    # Parsing an RFC 3339 duration with regular expressions is not
 
318
    # possible - there would have to be multiple places for the same
 
319
    # values, like seconds.  The current code, while more esoteric, is
 
320
    # cleaner without depending on a parsing library.  If Python had a
 
321
    # built-in library for parsing we would use it, but we'd like to
 
322
    # avoid excessive use of external libraries.
 
323
 
 
324
    # New type for defining tokens, syntax, and semantics all-in-one
 
325
    Token = collections.namedtuple("Token", (
 
326
        "regexp",  # To match token; if "value" is not None, must have
 
327
                   # a "group" containing digits
 
328
        "value",   # datetime.timedelta or None
 
329
        "followers"))           # Tokens valid after this token
 
330
    # RFC 3339 "duration" tokens, syntax, and semantics; taken from
 
331
    # the "duration" ABNF definition in RFC 3339, Appendix A.
 
332
    token_end = Token(re.compile(r"$"), None, frozenset())
 
333
    token_second = Token(re.compile(r"(\d+)S"),
 
334
                         datetime.timedelta(seconds=1),
 
335
                         frozenset((token_end, )))
 
336
    token_minute = Token(re.compile(r"(\d+)M"),
 
337
                         datetime.timedelta(minutes=1),
 
338
                         frozenset((token_second, token_end)))
 
339
    token_hour = Token(re.compile(r"(\d+)H"),
 
340
                       datetime.timedelta(hours=1),
 
341
                       frozenset((token_minute, token_end)))
 
342
    token_time = Token(re.compile(r"T"),
 
343
                       None,
 
344
                       frozenset((token_hour, token_minute,
 
345
                                  token_second)))
 
346
    token_day = Token(re.compile(r"(\d+)D"),
 
347
                      datetime.timedelta(days=1),
 
348
                      frozenset((token_time, token_end)))
 
349
    token_month = Token(re.compile(r"(\d+)M"),
 
350
                        datetime.timedelta(weeks=4),
 
351
                        frozenset((token_day, token_end)))
 
352
    token_year = Token(re.compile(r"(\d+)Y"),
 
353
                       datetime.timedelta(weeks=52),
 
354
                       frozenset((token_month, token_end)))
 
355
    token_week = Token(re.compile(r"(\d+)W"),
 
356
                       datetime.timedelta(weeks=1),
 
357
                       frozenset((token_end, )))
 
358
    token_duration = Token(re.compile(r"P"), None,
 
359
                           frozenset((token_year, token_month,
 
360
                                      token_day, token_time,
 
361
                                      token_week)))
 
362
    # Define starting values:
 
363
    # Value so far
 
364
    value = datetime.timedelta()
 
365
    found_token = None
 
366
    # Following valid tokens
 
367
    followers = frozenset((token_duration, ))
 
368
    # String left to parse
 
369
    s = duration
 
370
    # Loop until end token is found
 
371
    while found_token is not token_end:
 
372
        # Search for any currently valid tokens
 
373
        for token in followers:
 
374
            match = token.regexp.match(s)
 
375
            if match is not None:
 
376
                # Token found
 
377
                if token.value is not None:
 
378
                    # Value found, parse digits
 
379
                    factor = int(match.group(1), 10)
 
380
                    # Add to value so far
 
381
                    value += factor * token.value
 
382
                # Strip token from string
 
383
                s = token.regexp.sub("", s, 1)
 
384
                # Go to found token
 
385
                found_token = token
 
386
                # Set valid next tokens
 
387
                followers = found_token.followers
 
388
                break
 
389
        else:
 
390
            # No currently valid tokens were found
 
391
            raise ValueError("Invalid RFC 3339 duration: \"{}\""
 
392
                             .format(duration))
 
393
    # End token found
 
394
    return value
 
395
 
 
396
 
 
397
def parse_pre_1_6_1_interval(interval):
 
398
    r"""Parse an interval string as documented by Mandos before 1.6.1,
 
399
    and return a datetime.timedelta
 
400
 
 
401
    >>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)
 
402
    True
 
403
    >>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)
 
404
    True
 
405
    >>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)
 
406
    True
 
407
    >>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)
 
408
    True
 
409
    >>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)
 
410
    True
 
411
    >>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)
 
412
    True
 
413
    >>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)
 
414
    True
 
415
    >>> # Ignore unknown characters, allow any order and repetitions
 
416
    >>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \
 
417
    ... == datetime.timedelta(2, 480, 18000)
 
418
    True
 
419
 
 
420
    """
 
421
 
 
422
    value = datetime.timedelta(0)
 
423
    regexp = re.compile(r"(\d+)([dsmhw]?)")
 
424
 
 
425
    for num, suffix in regexp.findall(interval):
 
426
        if suffix == "d":
 
427
            value += datetime.timedelta(int(num))
 
428
        elif suffix == "s":
 
429
            value += datetime.timedelta(0, int(num))
 
430
        elif suffix == "m":
 
431
            value += datetime.timedelta(0, 0, 0, 0, int(num))
 
432
        elif suffix == "h":
 
433
            value += datetime.timedelta(0, 0, 0, 0, 0, int(num))
 
434
        elif suffix == "w":
 
435
            value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))
 
436
        elif suffix == "":
 
437
            value += datetime.timedelta(0, 0, 0, int(num))
 
438
    return value
 
439
 
 
440
 
 
441
def check_option_syntax(parser, options):
 
442
    """Apply additional restrictions on options, not expressible in
 
443
argparse"""
 
444
 
 
445
    def has_commands(options, commands=None):
 
446
        if commands is None:
 
447
            commands = (command.Enable,
 
448
                        command.Disable,
 
449
                        command.BumpTimeout,
 
450
                        command.StartChecker,
 
451
                        command.StopChecker,
 
452
                        command.IsEnabled,
 
453
                        command.Remove,
 
454
                        command.SetChecker,
 
455
                        command.SetTimeout,
 
456
                        command.SetExtendedTimeout,
 
457
                        command.SetInterval,
 
458
                        command.ApproveByDefault,
 
459
                        command.DenyByDefault,
 
460
                        command.SetApprovalDelay,
 
461
                        command.SetApprovalDuration,
 
462
                        command.SetHost,
 
463
                        command.SetSecret,
 
464
                        command.Approve,
 
465
                        command.Deny)
 
466
        return any(isinstance(cmd, commands)
 
467
                   for cmd in options.commands)
 
468
 
 
469
    if has_commands(options) and not (options.client or options.all):
215
470
        parser.error("Options require clients names or --all.")
216
 
    if options.verbose and has_actions(options):
217
 
        parser.error("--verbose can only be used alone or with"
218
 
                     " --all.")
219
 
    if options.all and not has_actions(options):
 
471
    if options.verbose and has_commands(options):
 
472
        parser.error("--verbose can only be used alone.")
 
473
    if (has_commands(options, (command.DumpJSON,))
 
474
        and (options.verbose or len(options.commands) > 1)):
 
475
        parser.error("--dump-json can only be used alone.")
 
476
    if options.all and not has_commands(options):
220
477
        parser.error("--all requires an action.")
221
 
    
222
 
    try:
223
 
        bus = dbus.SystemBus()
224
 
        mandos_dbus_objc = bus.get_object(busname, server_path)
225
 
    except dbus.exceptions.DBusException:
226
 
        print("Could not connect to Mandos server",
227
 
              file=sys.stderr)
228
 
        sys.exit(1)
229
 
    
230
 
    mandos_serv = dbus.Interface(mandos_dbus_objc,
231
 
                                 dbus_interface = server_interface)
232
 
    
233
 
    #block stderr since dbus library prints to stderr
234
 
    null = os.open(os.path.devnull, os.O_RDWR)
235
 
    stderrcopy = os.dup(sys.stderr.fileno())
236
 
    os.dup2(null, sys.stderr.fileno())
237
 
    os.close(null)
238
 
    try:
239
 
        try:
240
 
            mandos_clients = mandos_serv.GetAllClientsWithProperties()
241
 
        finally:
242
 
            #restore stderr
243
 
            os.dup2(stderrcopy, sys.stderr.fileno())
244
 
            os.close(stderrcopy)
245
 
    except dbus.exceptions.DBusException, e:
246
 
        print("Access denied: Accessing mandos server through dbus.",
247
 
              file=sys.stderr)
248
 
        sys.exit(1)
249
 
    
250
 
    # Compile dict of (clients: properties) to process
251
 
    clients={}
252
 
    
253
 
    if options.all or not options.client:
254
 
        clients = dict((bus.get_object(busname, path), properties)
255
 
                       for path, properties in
256
 
                       mandos_clients.iteritems())
257
 
    else:
258
 
        for name in options.client:
259
 
            for path, client in mandos_clients.iteritems():
260
 
                if client["Name"] == name:
261
 
                    client_objc = bus.get_object(busname, path)
262
 
                    clients[client_objc] = client
263
 
                    break
264
 
            else:
265
 
                print("Client not found on server: %r" % name,
266
 
                      file=sys.stderr)
267
 
                sys.exit(1)
268
 
    
269
 
    if not has_actions(options) and clients:
270
 
        if options.verbose:
271
 
            keywords = ("Name", "Enabled", "Timeout",
272
 
                        "LastCheckedOK", "Created", "Interval",
273
 
                        "Host", "Fingerprint", "CheckerRunning",
 
478
    if (has_commands(options, (command.IsEnabled,))
 
479
        and len(options.client) > 1):
 
480
        parser.error("--is-enabled requires exactly one client")
 
481
    if (len(options.commands) > 1
 
482
        and has_commands(options, (command.Remove,))
 
483
        and not has_commands(options, (command.Deny,))):
 
484
        parser.error("--remove can only be combined with --deny")
 
485
 
 
486
 
 
487
class dbus:
 
488
 
 
489
    class SystemBus:
 
490
 
 
491
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
 
492
 
 
493
        def get_managed_objects(self, busname, objectpath):
 
494
            return self.call_method("GetManagedObjects", busname,
 
495
                                    objectpath,
 
496
                                    self.object_manager_iface)
 
497
 
 
498
        properties_iface = "org.freedesktop.DBus.Properties"
 
499
 
 
500
        def set_property(self, busname, objectpath, interface, key,
 
501
                         value):
 
502
            self.call_method("Set", busname, objectpath,
 
503
                             self.properties_iface, interface, key,
 
504
                             value)
 
505
 
 
506
        def call_method(self, methodname, busname, objectpath,
 
507
                        interface, *args):
 
508
            raise NotImplementedError()
 
509
 
 
510
    class MandosBus(SystemBus):
 
511
        busname_domain = "se.recompile"
 
512
        busname = busname_domain + ".Mandos"
 
513
        server_path = "/"
 
514
        server_interface = busname_domain + ".Mandos"
 
515
        client_interface = busname_domain + ".Mandos.Client"
 
516
        del busname_domain
 
517
 
 
518
        def get_clients_and_properties(self):
 
519
            managed_objects = self.get_managed_objects(
 
520
                self.busname, self.server_path)
 
521
            return {objpath: properties[self.client_interface]
 
522
                    for objpath, properties in managed_objects.items()
 
523
                    if self.client_interface in properties}
 
524
 
 
525
        def set_client_property(self, objectpath, key, value):
 
526
            return self.set_property(self.busname, objectpath,
 
527
                                     self.client_interface, key,
 
528
                                     value)
 
529
 
 
530
        def call_client_method(self, objectpath, method, *args):
 
531
            return self.call_method(method, self.busname, objectpath,
 
532
                                    self.client_interface, *args)
 
533
 
 
534
        def call_server_method(self, method, *args):
 
535
            return self.call_method(method, self.busname,
 
536
                                    self.server_path,
 
537
                                    self.server_interface, *args)
 
538
 
 
539
    class Error(Exception):
 
540
        pass
 
541
 
 
542
    class ConnectFailed(Error):
 
543
        pass
 
544
 
 
545
 
 
546
class dbus_python_adapter:
 
547
 
 
548
    class SystemBus(dbus.MandosBus):
 
549
        """Use dbus-python"""
 
550
 
 
551
        def __init__(self, module=dbus_python):
 
552
            self.dbus_python = module
 
553
            self.bus = self.dbus_python.SystemBus()
 
554
 
 
555
        @contextlib.contextmanager
 
556
        def convert_exception(self, exception_class=dbus.Error):
 
557
            try:
 
558
                yield
 
559
            except self.dbus_python.exceptions.DBusException as e:
 
560
                # This does what "raise from" would do
 
561
                exc = exception_class(*e.args)
 
562
                exc.__cause__ = e
 
563
                raise exc
 
564
 
 
565
        def call_method(self, methodname, busname, objectpath,
 
566
                        interface, *args):
 
567
            proxy_object = self.get_object(busname, objectpath)
 
568
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
569
                      interface, methodname,
 
570
                      ", ".join(repr(a) for a in args))
 
571
            method = getattr(proxy_object, methodname)
 
572
            with self.convert_exception():
 
573
                with dbus_python_adapter.SilenceLogger(
 
574
                        "dbus.proxies"):
 
575
                    value = method(*args, dbus_interface=interface)
 
576
            return self.type_filter(value)
 
577
 
 
578
        def get_object(self, busname, objectpath):
 
579
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
580
                      busname, objectpath)
 
581
            with self.convert_exception(dbus.ConnectFailed):
 
582
                return self.bus.get_object(busname, objectpath)
 
583
 
 
584
        def type_filter(self, value):
 
585
            """Convert the most bothersome types to Python types"""
 
586
            if isinstance(value, self.dbus_python.Boolean):
 
587
                return bool(value)
 
588
            if isinstance(value, self.dbus_python.ObjectPath):
 
589
                return str(value)
 
590
            # Also recurse into dictionaries
 
591
            if isinstance(value, self.dbus_python.Dictionary):
 
592
                return {self.type_filter(key):
 
593
                        self.type_filter(subval)
 
594
                        for key, subval in value.items()}
 
595
            return value
 
596
 
 
597
        def set_client_property(self, objectpath, key, value):
 
598
            if key == "Secret":
 
599
                if not isinstance(value, bytes):
 
600
                    value = value.encode("utf-8")
 
601
                value = self.dbus_python.ByteArray(value)
 
602
            return self.set_property(self.busname, objectpath,
 
603
                                     self.client_interface, key,
 
604
                                     value)
 
605
 
 
606
    class SilenceLogger:
 
607
        "Simple context manager to silence a particular logger"
 
608
 
 
609
        def __init__(self, loggername):
 
610
            self.logger = logging.getLogger(loggername)
 
611
 
 
612
        def __enter__(self):
 
613
            self.logger.addFilter(self.nullfilter)
 
614
 
 
615
        class NullFilter(logging.Filter):
 
616
            def filter(self, record):
 
617
                return False
 
618
 
 
619
        nullfilter = NullFilter()
 
620
 
 
621
        def __exit__(self, exc_type, exc_val, exc_tb):
 
622
            self.logger.removeFilter(self.nullfilter)
 
623
 
 
624
    class CachingBus(SystemBus):
 
625
        """A caching layer for dbus_python_adapter.SystemBus"""
 
626
 
 
627
        def __init__(self, *args, **kwargs):
 
628
            self.object_cache = {}
 
629
            super(dbus_python_adapter.CachingBus,
 
630
                  self).__init__(*args, **kwargs)
 
631
 
 
632
        def get_object(self, busname, objectpath):
 
633
            try:
 
634
                return self.object_cache[(busname, objectpath)]
 
635
            except KeyError:
 
636
                new_object = super(
 
637
                    dbus_python_adapter.CachingBus,
 
638
                    self).get_object(busname, objectpath)
 
639
                self.object_cache[(busname, objectpath)] = new_object
 
640
                return new_object
 
641
 
 
642
 
 
643
class pydbus_adapter:
 
644
    class SystemBus(dbus.MandosBus):
 
645
        def __init__(self, module=pydbus):
 
646
            self.pydbus = module
 
647
            self.bus = self.pydbus.SystemBus()
 
648
 
 
649
        @contextlib.contextmanager
 
650
        def convert_exception(self, exception_class=dbus.Error):
 
651
            try:
 
652
                yield
 
653
            except gi.repository.GLib.Error as e:
 
654
                # This does what "raise from" would do
 
655
                exc = exception_class(*e.args)
 
656
                exc.__cause__ = e
 
657
                raise exc
 
658
 
 
659
        def call_method(self, methodname, busname, objectpath,
 
660
                        interface, *args):
 
661
            proxy_object = self.get(busname, objectpath)
 
662
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
663
                      interface, methodname,
 
664
                      ", ".join(repr(a) for a in args))
 
665
            method = getattr(proxy_object[interface], methodname)
 
666
            with self.convert_exception():
 
667
                return method(*args)
 
668
 
 
669
        def get(self, busname, objectpath):
 
670
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
671
                      busname, objectpath)
 
672
            with self.convert_exception(dbus.ConnectFailed):
 
673
                if sys.version_info.major <= 2:
 
674
                    with warnings.catch_warnings():
 
675
                        warnings.filterwarnings(
 
676
                            "ignore", "", DeprecationWarning,
 
677
                            r"^xml\.etree\.ElementTree$")
 
678
                        return self.bus.get(busname, objectpath)
 
679
                else:
 
680
                    return self.bus.get(busname, objectpath)
 
681
 
 
682
        def set_property(self, busname, objectpath, interface, key,
 
683
                         value):
 
684
            proxy_object = self.get(busname, objectpath)
 
685
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
686
                      objectpath, self.properties_iface, interface,
 
687
                      key, value)
 
688
            setattr(proxy_object[interface], key, value)
 
689
 
 
690
    class CachingBus(SystemBus):
 
691
        """A caching layer for pydbus_adapter.SystemBus"""
 
692
 
 
693
        def __init__(self, *args, **kwargs):
 
694
            self.object_cache = {}
 
695
            super(pydbus_adapter.CachingBus,
 
696
                  self).__init__(*args, **kwargs)
 
697
 
 
698
        def get(self, busname, objectpath):
 
699
            try:
 
700
                return self.object_cache[(busname, objectpath)]
 
701
            except KeyError:
 
702
                new_object = (super(pydbus_adapter.CachingBus, self)
 
703
                              .get(busname, objectpath))
 
704
                self.object_cache[(busname, objectpath)] = new_object
 
705
                return new_object
 
706
 
 
707
 
 
708
class dbussy_adapter:
 
709
    class SystemBus(dbus.SystemBus):
 
710
        """Use DBussy"""
 
711
 
 
712
        def __init__(self, dbussy, ravel):
 
713
            self.dbussy = dbussy
 
714
            self.ravel = ravel
 
715
            self.bus = ravel.system_bus()
 
716
 
 
717
        @contextlib.contextmanager
 
718
        def convert_exception(self, exception_class=dbus.Error):
 
719
            try:
 
720
                yield
 
721
            except self.dbussy.DBusError as e:
 
722
                # This does what "raise from" would do
 
723
                exc = exception_class(*e.args)
 
724
                exc.__cause__ = e
 
725
                raise exc
 
726
 
 
727
        def call_method(self, methodname, busname, objectpath,
 
728
                        interface, *args):
 
729
            proxy_object = self.get_object(busname, objectpath)
 
730
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
731
                      interface, methodname,
 
732
                      ", ".join(repr(a) for a in args))
 
733
            iface = proxy_object.get_interface(interface)
 
734
            method = getattr(iface, methodname)
 
735
            with self.convert_exception(dbus.Error):
 
736
                value = method(*args)
 
737
            # DBussy returns values either as an empty list or as a
 
738
            # list of one element with the return value
 
739
            if value:
 
740
                return self.type_filter(value[0])
 
741
 
 
742
        def get_object(self, busname, objectpath):
 
743
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
744
                      busname, objectpath)
 
745
            with self.convert_exception(dbus.ConnectFailed):
 
746
                return self.bus[busname][objectpath]
 
747
 
 
748
        def type_filter(self, value):
 
749
            """Convert the most bothersome types to Python types"""
 
750
            # A D-Bus Variant value is represented as the Python type
 
751
            # Tuple[dbussy.DBUS.Signature, Any]
 
752
            if isinstance(value, tuple):
 
753
                if (len(value) == 2
 
754
                    and isinstance(value[0],
 
755
                                   self.dbussy.DBUS.Signature)):
 
756
                    return self.type_filter(value[1])
 
757
            elif isinstance(value, self.dbussy.DBUS.ObjectPath):
 
758
                return str(value)
 
759
            # Also recurse into dictionaries
 
760
            elif isinstance(value, dict):
 
761
                return {self.type_filter(key):
 
762
                        self.type_filter(subval)
 
763
                        for key, subval in value.items()}
 
764
            return value
 
765
 
 
766
        def set_property(self, busname, objectpath, interface, key,
 
767
                         value):
 
768
            proxy_object = self.get_object(busname, objectpath)
 
769
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
770
                      objectpath, self.properties_iface, interface,
 
771
                      key, value)
 
772
            if key == "Secret":
 
773
                # DBussy wants a Byte Array to be a sequence of
 
774
                # values, not a byte string
 
775
                value = tuple(value)
 
776
            setattr(proxy_object.get_interface(interface), key, value)
 
777
 
 
778
    class MandosBus(SystemBus, dbus.MandosBus):
 
779
        pass
 
780
 
 
781
    class CachingBus(MandosBus):
 
782
        """A caching layer for dbussy_adapter.MandosBus"""
 
783
 
 
784
        def __init__(self, *args, **kwargs):
 
785
            self.object_cache = {}
 
786
            super(dbussy_adapter.CachingBus, self).__init__(*args,
 
787
                                                            **kwargs)
 
788
 
 
789
        def get_object(self, busname, objectpath):
 
790
            try:
 
791
                return self.object_cache[(busname, objectpath)]
 
792
            except KeyError:
 
793
                new_object = super(
 
794
                    dbussy_adapter.CachingBus,
 
795
                    self).get_object(busname, objectpath)
 
796
                self.object_cache[(busname, objectpath)] = new_object
 
797
                return new_object
 
798
 
 
799
 
 
800
def commands_from_options(options):
 
801
 
 
802
    commands = list(options.commands)
 
803
 
 
804
    def find_cmd(cmd, commands):
 
805
        i = 0
 
806
        for i, c in enumerate(commands):
 
807
            if isinstance(c, cmd):
 
808
                return i
 
809
        return i+1
 
810
 
 
811
    # If command.Remove is present, move any instances of command.Deny
 
812
    # to occur ahead of command.Remove.
 
813
    index_of_remove = find_cmd(command.Remove, commands)
 
814
    before_remove = commands[:index_of_remove]
 
815
    after_remove = commands[index_of_remove:]
 
816
    cleaned_after = []
 
817
    for cmd in after_remove:
 
818
        if isinstance(cmd, command.Deny):
 
819
            before_remove.append(cmd)
 
820
        else:
 
821
            cleaned_after.append(cmd)
 
822
    if cleaned_after != after_remove:
 
823
        commands = before_remove + cleaned_after
 
824
 
 
825
    # If no command option has been given, show table of clients,
 
826
    # optionally verbosely
 
827
    if not commands:
 
828
        commands.append(command.PrintTable(verbose=options.verbose))
 
829
 
 
830
    return commands
 
831
 
 
832
 
 
833
class command:
 
834
    """A namespace for command classes"""
 
835
 
 
836
    class Base:
 
837
        """Abstract base class for commands"""
 
838
 
 
839
        def run(self, clients, bus=None):
 
840
            """Normal commands should implement run_on_one_client(),
 
841
but commands which want to operate on all clients at the same time can
 
842
override this run() method instead.
 
843
"""
 
844
            self.bus = bus
 
845
            for client, properties in clients.items():
 
846
                self.run_on_one_client(client, properties)
 
847
 
 
848
    class IsEnabled(Base):
 
849
        def run(self, clients, bus=None):
 
850
            properties = next(iter(clients.values()))
 
851
            if properties["Enabled"]:
 
852
                sys.exit(0)
 
853
            sys.exit(1)
 
854
 
 
855
    class Approve(Base):
 
856
        def run_on_one_client(self, client, properties):
 
857
            self.bus.call_client_method(client, "Approve", True)
 
858
 
 
859
    class Deny(Base):
 
860
        def run_on_one_client(self, client, properties):
 
861
            self.bus.call_client_method(client, "Approve", False)
 
862
 
 
863
    class Remove(Base):
 
864
        def run(self, clients, bus):
 
865
            for clientpath in frozenset(clients.keys()):
 
866
                bus.call_server_method("RemoveClient", clientpath)
 
867
 
 
868
    class Output(Base):
 
869
        """Abstract class for commands outputting client details"""
 
870
        all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
 
871
                        "Created", "Interval", "Host", "KeyID",
 
872
                        "Fingerprint", "CheckerRunning",
274
873
                        "LastEnabled", "ApprovalPending",
275
 
                        "ApprovedByDefault",
276
 
                        "LastApprovalRequest", "ApprovalDelay",
277
 
                        "ApprovalDuration", "Checker",
278
 
                        "ExtendedTimeout")
279
 
        else:
280
 
            keywords = defaultkeywords
281
 
        
282
 
        print_clients(clients.values(), keywords)
283
 
    else:
284
 
        # Process each client in the list by all selected options
285
 
        for client in clients:
286
 
            if options.remove:
287
 
                mandos_serv.RemoveClient(client.__dbus_object_path__)
288
 
            if options.enable:
289
 
                client.Enable(dbus_interface=client_interface)
290
 
            if options.disable:
291
 
                client.Disable(dbus_interface=client_interface)
292
 
            if options.bump_timeout:
293
 
                client.CheckedOK(dbus_interface=client_interface)
294
 
            if options.start_checker:
295
 
                client.StartChecker(dbus_interface=client_interface)
296
 
            if options.stop_checker:
297
 
                client.StopChecker(dbus_interface=client_interface)
298
 
            if options.is_enabled:
299
 
                sys.exit(0 if client.Get(client_interface,
300
 
                                         "Enabled",
301
 
                                         dbus_interface=
302
 
                                         dbus.PROPERTIES_IFACE)
303
 
                         else 1)
304
 
            if options.checker is not None:
305
 
                client.Set(client_interface, "Checker",
306
 
                           options.checker,
307
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
308
 
            if options.host is not None:
309
 
                client.Set(client_interface, "Host", options.host,
310
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
311
 
            if options.interval is not None:
312
 
                client.Set(client_interface, "Interval",
313
 
                           timedelta_to_milliseconds
314
 
                           (string_to_delta(options.interval)),
315
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
316
 
            if options.approval_delay is not None:
317
 
                client.Set(client_interface, "ApprovalDelay",
318
 
                           timedelta_to_milliseconds
319
 
                           (string_to_delta(options.
320
 
                                            approval_delay)),
321
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
322
 
            if options.approval_duration is not None:
323
 
                client.Set(client_interface, "ApprovalDuration",
324
 
                           timedelta_to_milliseconds
325
 
                           (string_to_delta(options.
326
 
                                            approval_duration)),
327
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
328
 
            if options.timeout is not None:
329
 
                client.Set(client_interface, "Timeout",
330
 
                           timedelta_to_milliseconds
331
 
                           (string_to_delta(options.timeout)),
332
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
333
 
            if options.extended_timeout is not None:
334
 
                client.Set(client_interface, "ExtendedTimeout",
335
 
                           timedelta_to_milliseconds
336
 
                           (string_to_delta(options.extended_timeout)),
337
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
338
 
            if options.secret is not None:
339
 
                client.Set(client_interface, "Secret",
340
 
                           dbus.ByteArray(open(options.secret,
341
 
                                               "rb").read()),
342
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
343
 
            if options.approved_by_default is not None:
344
 
                client.Set(client_interface, "ApprovedByDefault",
345
 
                           dbus.Boolean(options
346
 
                                        .approved_by_default),
347
 
                           dbus_interface=dbus.PROPERTIES_IFACE)
348
 
            if options.approve:
349
 
                client.Approve(dbus.Boolean(True),
350
 
                               dbus_interface=client_interface)
351
 
            elif options.deny:
352
 
                client.Approve(dbus.Boolean(False),
353
 
                               dbus_interface=client_interface)
 
874
                        "ApprovedByDefault", "LastApprovalRequest",
 
875
                        "ApprovalDelay", "ApprovalDuration",
 
876
                        "Checker", "ExtendedTimeout", "Expires",
 
877
                        "LastCheckerStatus")
 
878
 
 
879
    class DumpJSON(Output):
 
880
        def run(self, clients, bus=None):
 
881
            data = {properties["Name"]:
 
882
                    {key: properties[key]
 
883
                     for key in self.all_keywords}
 
884
                    for properties in clients.values()}
 
885
            print(json.dumps(data, indent=4, separators=(",", ": ")))
 
886
 
 
887
    class PrintTable(Output):
 
888
        def __init__(self, verbose=False):
 
889
            self.verbose = verbose
 
890
 
 
891
        def run(self, clients, bus=None):
 
892
            default_keywords = ("Name", "Enabled", "Timeout",
 
893
                                "LastCheckedOK")
 
894
            keywords = default_keywords
 
895
            if self.verbose:
 
896
                keywords = self.all_keywords
 
897
            print(self.TableOfClients(clients.values(), keywords))
 
898
 
 
899
        class TableOfClients:
 
900
            tableheaders = {
 
901
                "Name": "Name",
 
902
                "Enabled": "Enabled",
 
903
                "Timeout": "Timeout",
 
904
                "LastCheckedOK": "Last Successful Check",
 
905
                "LastApprovalRequest": "Last Approval Request",
 
906
                "Created": "Created",
 
907
                "Interval": "Interval",
 
908
                "Host": "Host",
 
909
                "Fingerprint": "Fingerprint",
 
910
                "KeyID": "Key ID",
 
911
                "CheckerRunning": "Check Is Running",
 
912
                "LastEnabled": "Last Enabled",
 
913
                "ApprovalPending": "Approval Is Pending",
 
914
                "ApprovedByDefault": "Approved By Default",
 
915
                "ApprovalDelay": "Approval Delay",
 
916
                "ApprovalDuration": "Approval Duration",
 
917
                "Checker": "Checker",
 
918
                "ExtendedTimeout": "Extended Timeout",
 
919
                "Expires": "Expires",
 
920
                "LastCheckerStatus": "Last Checker Status",
 
921
            }
 
922
 
 
923
            def __init__(self, clients, keywords):
 
924
                self.clients = clients
 
925
                self.keywords = keywords
 
926
 
 
927
            def __str__(self):
 
928
                return "\n".join(self.rows())
 
929
 
 
930
            if sys.version_info.major == 2:
 
931
                __unicode__ = __str__
 
932
 
 
933
                def __str__(self):
 
934
                    return str(self).encode(
 
935
                        locale.getpreferredencoding())
 
936
 
 
937
            def rows(self):
 
938
                format_string = self.row_formatting_string()
 
939
                rows = [self.header_line(format_string)]
 
940
                rows.extend(self.client_line(client, format_string)
 
941
                            for client in self.clients)
 
942
                return rows
 
943
 
 
944
            def row_formatting_string(self):
 
945
                "Format string used to format table rows"
 
946
                return " ".join("{{{key}:{width}}}".format(
 
947
                    width=max(len(self.tableheaders[key]),
 
948
                              *(len(self.string_from_client(client,
 
949
                                                            key))
 
950
                                for client in self.clients)),
 
951
                    key=key)
 
952
                                for key in self.keywords)
 
953
 
 
954
            def string_from_client(self, client, key):
 
955
                return self.valuetostring(client[key], key)
 
956
 
 
957
            @classmethod
 
958
            def valuetostring(cls, value, keyword):
 
959
                if isinstance(value, bool):
 
960
                    return "Yes" if value else "No"
 
961
                if keyword in ("Timeout", "Interval", "ApprovalDelay",
 
962
                               "ApprovalDuration", "ExtendedTimeout"):
 
963
                    return cls.milliseconds_to_string(value)
 
964
                return str(value)
 
965
 
 
966
            def header_line(self, format_string):
 
967
                return format_string.format(**self.tableheaders)
 
968
 
 
969
            def client_line(self, client, format_string):
 
970
                return format_string.format(
 
971
                    **{key: self.string_from_client(client, key)
 
972
                       for key in self.keywords})
 
973
 
 
974
            @staticmethod
 
975
            def milliseconds_to_string(ms):
 
976
                td = datetime.timedelta(0, 0, 0, ms)
 
977
                return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
 
978
                        .format(days="{}T".format(td.days)
 
979
                                if td.days else "",
 
980
                                hours=td.seconds // 3600,
 
981
                                minutes=(td.seconds % 3600) // 60,
 
982
                                seconds=td.seconds % 60))
 
983
 
 
984
    class PropertySetter(Base):
 
985
        "Abstract class for Actions for setting one client property"
 
986
 
 
987
        def run_on_one_client(self, client, properties=None):
 
988
            """Set the Client's D-Bus property"""
 
989
            self.bus.set_client_property(client, self.propname,
 
990
                                         self.value_to_set)
 
991
 
 
992
        @property
 
993
        def propname(self):
 
994
            raise NotImplementedError()
 
995
 
 
996
    class Enable(PropertySetter):
 
997
        propname = "Enabled"
 
998
        value_to_set = True
 
999
 
 
1000
    class Disable(PropertySetter):
 
1001
        propname = "Enabled"
 
1002
        value_to_set = False
 
1003
 
 
1004
    class BumpTimeout(PropertySetter):
 
1005
        propname = "LastCheckedOK"
 
1006
        value_to_set = ""
 
1007
 
 
1008
    class StartChecker(PropertySetter):
 
1009
        propname = "CheckerRunning"
 
1010
        value_to_set = True
 
1011
 
 
1012
    class StopChecker(PropertySetter):
 
1013
        propname = "CheckerRunning"
 
1014
        value_to_set = False
 
1015
 
 
1016
    class ApproveByDefault(PropertySetter):
 
1017
        propname = "ApprovedByDefault"
 
1018
        value_to_set = True
 
1019
 
 
1020
    class DenyByDefault(PropertySetter):
 
1021
        propname = "ApprovedByDefault"
 
1022
        value_to_set = False
 
1023
 
 
1024
    class PropertySetterValue(PropertySetter):
 
1025
        """Abstract class for PropertySetter recieving a value as
 
1026
constructor argument instead of a class attribute."""
 
1027
 
 
1028
        def __init__(self, value):
 
1029
            self.value_to_set = value
 
1030
 
 
1031
        @classmethod
 
1032
        def argparse(cls, argtype):
 
1033
            def cmdtype(arg):
 
1034
                return cls(argtype(arg))
 
1035
            return cmdtype
 
1036
 
 
1037
    class SetChecker(PropertySetterValue):
 
1038
        propname = "Checker"
 
1039
 
 
1040
    class SetHost(PropertySetterValue):
 
1041
        propname = "Host"
 
1042
 
 
1043
    class SetSecret(PropertySetterValue):
 
1044
        propname = "Secret"
 
1045
 
 
1046
        @property
 
1047
        def value_to_set(self):
 
1048
            return self._vts
 
1049
 
 
1050
        @value_to_set.setter
 
1051
        def value_to_set(self, value):
 
1052
            """When setting, read data from supplied file object"""
 
1053
            self._vts = value.read()
 
1054
            value.close()
 
1055
 
 
1056
    class PropertySetterValueMilliseconds(PropertySetterValue):
 
1057
        """Abstract class for PropertySetterValue taking a value
 
1058
argument as a datetime.timedelta() but should store it as
 
1059
milliseconds."""
 
1060
 
 
1061
        @property
 
1062
        def value_to_set(self):
 
1063
            return self._vts
 
1064
 
 
1065
        @value_to_set.setter
 
1066
        def value_to_set(self, value):
 
1067
            "When setting, convert value from a datetime.timedelta"
 
1068
            self._vts = int(round(value.total_seconds() * 1000))
 
1069
 
 
1070
    class SetTimeout(PropertySetterValueMilliseconds):
 
1071
        propname = "Timeout"
 
1072
 
 
1073
    class SetExtendedTimeout(PropertySetterValueMilliseconds):
 
1074
        propname = "ExtendedTimeout"
 
1075
 
 
1076
    class SetInterval(PropertySetterValueMilliseconds):
 
1077
        propname = "Interval"
 
1078
 
 
1079
    class SetApprovalDelay(PropertySetterValueMilliseconds):
 
1080
        propname = "ApprovalDelay"
 
1081
 
 
1082
    class SetApprovalDuration(PropertySetterValueMilliseconds):
 
1083
        propname = "ApprovalDuration"
 
1084
 
 
1085
 
 
1086
 
 
1087
class TestCaseWithAssertLogs(unittest.TestCase):
 
1088
    """unittest.TestCase.assertLogs only exists in Python 3.4"""
 
1089
 
 
1090
    if not hasattr(unittest.TestCase, "assertLogs"):
 
1091
        @contextlib.contextmanager
 
1092
        def assertLogs(self, logger, level=logging.INFO):
 
1093
            capturing_handler = self.CapturingLevelHandler(level)
 
1094
            old_level = logger.level
 
1095
            old_propagate = logger.propagate
 
1096
            logger.addHandler(capturing_handler)
 
1097
            logger.setLevel(level)
 
1098
            logger.propagate = False
 
1099
            try:
 
1100
                yield capturing_handler.watcher
 
1101
            finally:
 
1102
                logger.propagate = old_propagate
 
1103
                logger.removeHandler(capturing_handler)
 
1104
                logger.setLevel(old_level)
 
1105
            self.assertGreater(len(capturing_handler.watcher.records),
 
1106
                               0)
 
1107
 
 
1108
        class CapturingLevelHandler(logging.Handler):
 
1109
            def __init__(self, level, *args, **kwargs):
 
1110
                logging.Handler.__init__(self, *args, **kwargs)
 
1111
                self.watcher = self.LoggingWatcher([], [])
 
1112
            def emit(self, record):
 
1113
                self.watcher.records.append(record)
 
1114
                self.watcher.output.append(self.format(record))
 
1115
 
 
1116
            LoggingWatcher = collections.namedtuple("LoggingWatcher",
 
1117
                                                    ("records",
 
1118
                                                     "output"))
 
1119
 
 
1120
 
 
1121
class Unique:
 
1122
    """Class for objects which exist only to be unique objects, since
 
1123
unittest.mock.sentinel only exists in Python 3.3"""
 
1124
 
 
1125
 
 
1126
class Test_string_to_delta(TestCaseWithAssertLogs):
 
1127
    # Just test basic RFC 3339 functionality here, the doc string for
 
1128
    # rfc3339_duration_to_delta() already has more comprehensive
 
1129
    # tests, which are run by doctest.
 
1130
 
 
1131
    def test_rfc3339_zero_seconds(self):
 
1132
        self.assertEqual(datetime.timedelta(),
 
1133
                         string_to_delta("PT0S"))
 
1134
 
 
1135
    def test_rfc3339_zero_days(self):
 
1136
        self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
 
1137
 
 
1138
    def test_rfc3339_one_second(self):
 
1139
        self.assertEqual(datetime.timedelta(0, 1),
 
1140
                         string_to_delta("PT1S"))
 
1141
 
 
1142
    def test_rfc3339_two_hours(self):
 
1143
        self.assertEqual(datetime.timedelta(0, 7200),
 
1144
                         string_to_delta("PT2H"))
 
1145
 
 
1146
    def test_falls_back_to_pre_1_6_1_with_warning(self):
 
1147
        with self.assertLogs(log, logging.WARNING):
 
1148
            value = string_to_delta("2h")
 
1149
        self.assertEqual(datetime.timedelta(0, 7200), value)
 
1150
 
 
1151
 
 
1152
class Test_check_option_syntax(unittest.TestCase):
 
1153
    def setUp(self):
 
1154
        self.parser = argparse.ArgumentParser()
 
1155
        add_command_line_options(self.parser)
 
1156
 
 
1157
    def test_actions_requires_client_or_all(self):
 
1158
        for action, value in self.actions.items():
 
1159
            args = self.actionargs(action, value)
 
1160
            with self.assertParseError():
 
1161
                self.parse_args(args)
 
1162
 
 
1163
    # This mostly corresponds to the definition from has_commands() in
 
1164
    # check_option_syntax()
 
1165
    actions = {
 
1166
        "--enable": None,
 
1167
        "--disable": None,
 
1168
        "--bump-timeout": None,
 
1169
        "--start-checker": None,
 
1170
        "--stop-checker": None,
 
1171
        "--is-enabled": None,
 
1172
        "--remove": None,
 
1173
        "--checker": "x",
 
1174
        "--timeout": "PT0S",
 
1175
        "--extended-timeout": "PT0S",
 
1176
        "--interval": "PT0S",
 
1177
        "--approve-by-default": None,
 
1178
        "--deny-by-default": None,
 
1179
        "--approval-delay": "PT0S",
 
1180
        "--approval-duration": "PT0S",
 
1181
        "--host": "hostname",
 
1182
        "--secret": "/dev/null",
 
1183
        "--approve": None,
 
1184
        "--deny": None,
 
1185
    }
 
1186
 
 
1187
    @staticmethod
 
1188
    def actionargs(action, value, *args):
 
1189
        if value is not None:
 
1190
            return [action, value] + list(args)
 
1191
        else:
 
1192
            return [action] + list(args)
 
1193
 
 
1194
    @contextlib.contextmanager
 
1195
    def assertParseError(self):
 
1196
        with self.assertRaises(SystemExit) as e:
 
1197
            with self.redirect_stderr_to_devnull():
 
1198
                yield
 
1199
        # Exit code from argparse is guaranteed to be "2".  Reference:
 
1200
        # https://docs.python.org/3/library
 
1201
        # /argparse.html#exiting-methods
 
1202
        self.assertEqual(2, e.exception.code)
 
1203
 
 
1204
    def parse_args(self, args):
 
1205
        options = self.parser.parse_args(args)
 
1206
        check_option_syntax(self.parser, options)
 
1207
 
 
1208
    @staticmethod
 
1209
    @contextlib.contextmanager
 
1210
    def redirect_stderr_to_devnull():
 
1211
        old_stderr = sys.stderr
 
1212
        with contextlib.closing(open(os.devnull, "w")) as null:
 
1213
            sys.stderr = null
 
1214
            try:
 
1215
                yield
 
1216
            finally:
 
1217
                sys.stderr = old_stderr
 
1218
 
 
1219
    def check_option_syntax(self, options):
 
1220
        check_option_syntax(self.parser, options)
 
1221
 
 
1222
    def test_actions_all_conflicts_with_verbose(self):
 
1223
        for action, value in self.actions.items():
 
1224
            args = self.actionargs(action, value, "--all",
 
1225
                                   "--verbose")
 
1226
            with self.assertParseError():
 
1227
                self.parse_args(args)
 
1228
 
 
1229
    def test_actions_with_client_conflicts_with_verbose(self):
 
1230
        for action, value in self.actions.items():
 
1231
            args = self.actionargs(action, value, "--verbose",
 
1232
                                   "client")
 
1233
            with self.assertParseError():
 
1234
                self.parse_args(args)
 
1235
 
 
1236
    def test_dump_json_conflicts_with_verbose(self):
 
1237
        args = ["--dump-json", "--verbose"]
 
1238
        with self.assertParseError():
 
1239
            self.parse_args(args)
 
1240
 
 
1241
    def test_dump_json_conflicts_with_action(self):
 
1242
        for action, value in self.actions.items():
 
1243
            args = self.actionargs(action, value, "--dump-json")
 
1244
            with self.assertParseError():
 
1245
                self.parse_args(args)
 
1246
 
 
1247
    def test_all_can_not_be_alone(self):
 
1248
        args = ["--all"]
 
1249
        with self.assertParseError():
 
1250
            self.parse_args(args)
 
1251
 
 
1252
    def test_all_is_ok_with_any_action(self):
 
1253
        for action, value in self.actions.items():
 
1254
            args = self.actionargs(action, value, "--all")
 
1255
            self.parse_args(args)
 
1256
 
 
1257
    def test_any_action_is_ok_with_one_client(self):
 
1258
        for action, value in self.actions.items():
 
1259
            args = self.actionargs(action, value, "client")
 
1260
            self.parse_args(args)
 
1261
 
 
1262
    def test_one_client_with_all_actions_except_is_enabled(self):
 
1263
        for action, value in self.actions.items():
 
1264
            if action == "--is-enabled":
 
1265
                continue
 
1266
            args = self.actionargs(action, value, "client")
 
1267
            self.parse_args(args)
 
1268
 
 
1269
    def test_two_clients_with_all_actions_except_is_enabled(self):
 
1270
        for action, value in self.actions.items():
 
1271
            if action == "--is-enabled":
 
1272
                continue
 
1273
            args = self.actionargs(action, value, "client1",
 
1274
                                   "client2")
 
1275
            self.parse_args(args)
 
1276
 
 
1277
    def test_two_clients_are_ok_with_actions_except_is_enabled(self):
 
1278
        for action, value in self.actions.items():
 
1279
            if action == "--is-enabled":
 
1280
                continue
 
1281
            args = self.actionargs(action, value, "client1",
 
1282
                                   "client2")
 
1283
            self.parse_args(args)
 
1284
 
 
1285
    def test_is_enabled_fails_without_client(self):
 
1286
        args = ["--is-enabled"]
 
1287
        with self.assertParseError():
 
1288
            self.parse_args(args)
 
1289
 
 
1290
    def test_is_enabled_fails_with_two_clients(self):
 
1291
        args = ["--is-enabled", "client1", "client2"]
 
1292
        with self.assertParseError():
 
1293
            self.parse_args(args)
 
1294
 
 
1295
    def test_remove_can_only_be_combined_with_action_deny(self):
 
1296
        for action, value in self.actions.items():
 
1297
            if action in {"--remove", "--deny"}:
 
1298
                continue
 
1299
            args = self.actionargs(action, value, "--all",
 
1300
                                   "--remove")
 
1301
            with self.assertParseError():
 
1302
                self.parse_args(args)
 
1303
 
 
1304
 
 
1305
class Test_dbus_exceptions(unittest.TestCase):
 
1306
 
 
1307
    def test_dbus_ConnectFailed_is_Error(self):
 
1308
        with self.assertRaises(dbus.Error):
 
1309
            raise dbus.ConnectFailed()
 
1310
 
 
1311
 
 
1312
class Test_dbus_MandosBus(unittest.TestCase):
 
1313
 
 
1314
    class MockMandosBus(dbus.MandosBus):
 
1315
        def __init__(self):
 
1316
            self._name = "se.recompile.Mandos"
 
1317
            self._server_path = "/"
 
1318
            self._server_interface = "se.recompile.Mandos"
 
1319
            self._client_interface = "se.recompile.Mandos.Client"
 
1320
            self.calls = []
 
1321
            self.call_method_return = Unique()
 
1322
 
 
1323
        def call_method(self, methodname, busname, objectpath,
 
1324
                        interface, *args):
 
1325
            self.calls.append((methodname, busname, objectpath,
 
1326
                               interface, args))
 
1327
            return self.call_method_return
 
1328
 
 
1329
    def setUp(self):
 
1330
        self.bus = self.MockMandosBus()
 
1331
 
 
1332
    def test_set_client_property(self):
 
1333
        self.bus.set_client_property("objectpath", "key", "value")
 
1334
        expected_call = ("Set", self.bus._name, "objectpath",
 
1335
                         "org.freedesktop.DBus.Properties",
 
1336
                         (self.bus._client_interface, "key", "value"))
 
1337
        self.assertIn(expected_call, self.bus.calls)
 
1338
 
 
1339
    def test_call_client_method(self):
 
1340
        ret = self.bus.call_client_method("objectpath", "methodname")
 
1341
        self.assertIs(self.bus.call_method_return, ret)
 
1342
        expected_call = ("methodname", self.bus._name, "objectpath",
 
1343
                         self.bus._client_interface, ())
 
1344
        self.assertIn(expected_call, self.bus.calls)
 
1345
 
 
1346
    def test_call_client_method_with_args(self):
 
1347
        args = (Unique(), Unique())
 
1348
        ret = self.bus.call_client_method("objectpath", "methodname",
 
1349
                                          *args)
 
1350
        self.assertIs(self.bus.call_method_return, ret)
 
1351
        expected_call = ("methodname", self.bus._name, "objectpath",
 
1352
                         self.bus._client_interface,
 
1353
                         (args[0], args[1]))
 
1354
        self.assertIn(expected_call, self.bus.calls)
 
1355
 
 
1356
    def test_get_clients_and_properties(self):
 
1357
        managed_objects = {
 
1358
            "objectpath": {
 
1359
                self.bus._client_interface: {
 
1360
                    "key": "value",
 
1361
                    "bool": True,
 
1362
                },
 
1363
                "irrelevant_interface": {
 
1364
                    "key": "othervalue",
 
1365
                    "bool": False,
 
1366
                },
 
1367
            },
 
1368
            "other_objectpath": {
 
1369
                "other_irrelevant_interface": {
 
1370
                    "key": "value 3",
 
1371
                    "bool": None,
 
1372
                },
 
1373
            },
 
1374
        }
 
1375
        expected_clients_and_properties = {
 
1376
            "objectpath": {
 
1377
                "key": "value",
 
1378
                "bool": True,
 
1379
            }
 
1380
        }
 
1381
        self.bus.call_method_return = managed_objects
 
1382
        ret = self.bus.get_clients_and_properties()
 
1383
        self.assertDictEqual(expected_clients_and_properties, ret)
 
1384
        expected_call = ("GetManagedObjects", self.bus._name,
 
1385
                         self.bus._server_path,
 
1386
                         "org.freedesktop.DBus.ObjectManager", ())
 
1387
        self.assertIn(expected_call, self.bus.calls)
 
1388
 
 
1389
    def test_call_server_method(self):
 
1390
        ret = self.bus.call_server_method("methodname")
 
1391
        self.assertIs(self.bus.call_method_return, ret)
 
1392
        expected_call = ("methodname", self.bus._name,
 
1393
                         self.bus._server_path,
 
1394
                         self.bus._server_interface, ())
 
1395
        self.assertIn(expected_call, self.bus.calls)
 
1396
 
 
1397
    def test_call_server_method_with_args(self):
 
1398
        args = (Unique(), Unique())
 
1399
        ret = self.bus.call_server_method("methodname", *args)
 
1400
        self.assertIs(self.bus.call_method_return, ret)
 
1401
        expected_call = ("methodname", self.bus._name,
 
1402
                         self.bus._server_path,
 
1403
                         self.bus._server_interface,
 
1404
                         (args[0], args[1]))
 
1405
        self.assertIn(expected_call, self.bus.calls)
 
1406
 
 
1407
 
 
1408
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
 
1409
 
 
1410
    def MockDBusPython_func(self, func):
 
1411
        class mock_dbus_python:
 
1412
            """mock dbus-python module"""
 
1413
            class exceptions:
 
1414
                """Pseudo-namespace"""
 
1415
                class DBusException(Exception):
 
1416
                    pass
 
1417
            class SystemBus:
 
1418
                @staticmethod
 
1419
                def get_object(busname, objectpath):
 
1420
                    DBusObject = collections.namedtuple(
 
1421
                        "DBusObject", ("methodname", "Set"))
 
1422
                    def method(*args, **kwargs):
 
1423
                        self.assertEqual({"dbus_interface":
 
1424
                                          "interface"},
 
1425
                                         kwargs)
 
1426
                        return func(*args)
 
1427
                    def set_property(interface, key, value,
 
1428
                                     dbus_interface=None):
 
1429
                        self.assertEqual(
 
1430
                            "org.freedesktop.DBus.Properties",
 
1431
                            dbus_interface)
 
1432
                        self.assertEqual("Secret", key)
 
1433
                        return func(interface, key, value,
 
1434
                                    dbus_interface=dbus_interface)
 
1435
                    return DBusObject(methodname=method,
 
1436
                                      Set=set_property)
 
1437
            class Boolean:
 
1438
                def __init__(self, value):
 
1439
                    self.value = bool(value)
 
1440
                def __bool__(self):
 
1441
                    return self.value
 
1442
                if sys.version_info.major == 2:
 
1443
                    __nonzero__ = __bool__
 
1444
            class ObjectPath(str):
 
1445
                pass
 
1446
            class Dictionary(dict):
 
1447
                pass
 
1448
            class ByteArray(bytes):
 
1449
                pass
 
1450
        return mock_dbus_python
 
1451
 
 
1452
    def call_method(self, bus, methodname, busname, objectpath,
 
1453
                    interface, *args):
 
1454
        with self.assertLogs(log, logging.DEBUG):
 
1455
            return bus.call_method(methodname, busname, objectpath,
 
1456
                                   interface, *args)
 
1457
 
 
1458
    def test_call_method_returns(self):
 
1459
        expected_method_return = Unique()
 
1460
        method_args = (Unique(), Unique())
 
1461
        def func(*args):
 
1462
            self.assertEqual(len(method_args), len(args))
 
1463
            for marg, arg in zip(method_args, args):
 
1464
                self.assertIs(marg, arg)
 
1465
            return expected_method_return
 
1466
        mock_dbus_python = self.MockDBusPython_func(func)
 
1467
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1468
        ret = self.call_method(bus, "methodname", "busname",
 
1469
                               "objectpath", "interface",
 
1470
                               *method_args)
 
1471
        self.assertIs(ret, expected_method_return)
 
1472
 
 
1473
    def test_call_method_filters_bool_true(self):
 
1474
        def func():
 
1475
            return method_return
 
1476
        mock_dbus_python = self.MockDBusPython_func(func)
 
1477
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1478
        method_return = mock_dbus_python.Boolean(True)
 
1479
        ret = self.call_method(bus, "methodname", "busname",
 
1480
                               "objectpath", "interface")
 
1481
        self.assertTrue(ret)
 
1482
        self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
 
1483
 
 
1484
    def test_call_method_filters_bool_false(self):
 
1485
        def func():
 
1486
            return method_return
 
1487
        mock_dbus_python = self.MockDBusPython_func(func)
 
1488
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1489
        method_return = mock_dbus_python.Boolean(False)
 
1490
        ret = self.call_method(bus, "methodname", "busname",
 
1491
                               "objectpath", "interface")
 
1492
        self.assertFalse(ret)
 
1493
        self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
 
1494
 
 
1495
    def test_call_method_filters_objectpath(self):
 
1496
        def func():
 
1497
            return method_return
 
1498
        mock_dbus_python = self.MockDBusPython_func(func)
 
1499
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1500
        method_return = mock_dbus_python.ObjectPath("objectpath")
 
1501
        ret = self.call_method(bus, "methodname", "busname",
 
1502
                               "objectpath", "interface")
 
1503
        self.assertEqual("objectpath", ret)
 
1504
        self.assertIsNot("objectpath", ret)
 
1505
        self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
 
1506
 
 
1507
    def test_call_method_filters_booleans_in_dict(self):
 
1508
        def func():
 
1509
            return method_return
 
1510
        mock_dbus_python = self.MockDBusPython_func(func)
 
1511
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1512
        method_return = mock_dbus_python.Dictionary(
 
1513
        {mock_dbus_python.Boolean(True):
 
1514
         mock_dbus_python.Boolean(False),
 
1515
         mock_dbus_python.Boolean(False):
 
1516
         mock_dbus_python.Boolean(True)})
 
1517
        ret = self.call_method(bus, "methodname", "busname",
 
1518
                               "objectpath", "interface")
 
1519
        expected_method_return = {True: False,
 
1520
                                  False: True}
 
1521
        self.assertEqual(expected_method_return, ret)
 
1522
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1523
 
 
1524
    def test_call_method_filters_objectpaths_in_dict(self):
 
1525
        def func():
 
1526
            return method_return
 
1527
        mock_dbus_python = self.MockDBusPython_func(func)
 
1528
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1529
        method_return = mock_dbus_python.Dictionary(
 
1530
        {mock_dbus_python.ObjectPath("objectpath_key_1"):
 
1531
         mock_dbus_python.ObjectPath("objectpath_value_1"),
 
1532
         mock_dbus_python.ObjectPath("objectpath_key_2"):
 
1533
         mock_dbus_python.ObjectPath("objectpath_value_2")})
 
1534
        ret = self.call_method(bus, "methodname", "busname",
 
1535
                               "objectpath", "interface")
 
1536
        expected_method_return = {str(key): str(value)
 
1537
                                  for key, value in
 
1538
                                  method_return.items()}
 
1539
        self.assertEqual(expected_method_return, ret)
 
1540
        self.assertIsInstance(ret, dict)
 
1541
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1542
 
 
1543
    def test_call_method_filters_dict_in_dict(self):
 
1544
        def func():
 
1545
            return method_return
 
1546
        mock_dbus_python = self.MockDBusPython_func(func)
 
1547
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1548
        method_return = mock_dbus_python.Dictionary(
 
1549
        {"key1": mock_dbus_python.Dictionary({"key11": "value11",
 
1550
                                              "key12": "value12"}),
 
1551
         "key2": mock_dbus_python.Dictionary({"key21": "value21",
 
1552
                                              "key22": "value22"})})
 
1553
        ret = self.call_method(bus, "methodname", "busname",
 
1554
                               "objectpath", "interface")
 
1555
        expected_method_return = {
 
1556
            "key1": {"key11": "value11",
 
1557
                     "key12": "value12"},
 
1558
            "key2": {"key21": "value21",
 
1559
                     "key22": "value22"},
 
1560
        }
 
1561
        self.assertEqual(expected_method_return, ret)
 
1562
        self.assertIsInstance(ret, dict)
 
1563
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1564
        for key, value in ret.items():
 
1565
            self.assertIsInstance(value, dict)
 
1566
            self.assertEqual(expected_method_return[key], value)
 
1567
            self.assertNotIsInstance(value,
 
1568
                                     mock_dbus_python.Dictionary)
 
1569
 
 
1570
    def test_call_method_filters_dict_three_deep(self):
 
1571
        def func():
 
1572
            return method_return
 
1573
        mock_dbus_python = self.MockDBusPython_func(func)
 
1574
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1575
        method_return = mock_dbus_python.Dictionary(
 
1576
            {"key1":
 
1577
             mock_dbus_python.Dictionary(
 
1578
                 {"key2":
 
1579
                  mock_dbus_python.Dictionary(
 
1580
                      {"key3":
 
1581
                       mock_dbus_python.Boolean(True),
 
1582
                       }),
 
1583
                  }),
 
1584
             })
 
1585
        ret = self.call_method(bus, "methodname", "busname",
 
1586
                               "objectpath", "interface")
 
1587
        expected_method_return = {"key1": {"key2": {"key3": True}}}
 
1588
        self.assertEqual(expected_method_return, ret)
 
1589
        self.assertIsInstance(ret, dict)
 
1590
        self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
 
1591
        self.assertIsInstance(ret["key1"], dict)
 
1592
        self.assertNotIsInstance(ret["key1"],
 
1593
                                 mock_dbus_python.Dictionary)
 
1594
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
1595
        self.assertNotIsInstance(ret["key1"]["key2"],
 
1596
                                 mock_dbus_python.Dictionary)
 
1597
        self.assertTrue(ret["key1"]["key2"]["key3"])
 
1598
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
1599
                                 mock_dbus_python.Boolean)
 
1600
 
 
1601
    def test_call_method_handles_exception(self):
 
1602
        dbus_logger = logging.getLogger("dbus.proxies")
 
1603
 
 
1604
        def func():
 
1605
            dbus_logger.error("Test")
 
1606
            raise mock_dbus_python.exceptions.DBusException()
 
1607
 
 
1608
        mock_dbus_python = self.MockDBusPython_func(func)
 
1609
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1610
 
 
1611
        class CountingHandler(logging.Handler):
 
1612
            count = 0
 
1613
            def emit(self, record):
 
1614
                self.count += 1
 
1615
 
 
1616
        counting_handler = CountingHandler()
 
1617
 
 
1618
        dbus_logger.addHandler(counting_handler)
 
1619
 
 
1620
        try:
 
1621
            with self.assertRaises(dbus.Error) as e:
 
1622
                self.call_method(bus, "methodname", "busname",
 
1623
                                 "objectpath", "interface")
 
1624
        finally:
 
1625
            dbus_logger.removeFilter(counting_handler)
 
1626
 
 
1627
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1628
 
 
1629
        # Make sure the dbus logger was suppressed
 
1630
        self.assertEqual(0, counting_handler.count)
 
1631
 
 
1632
    def test_Set_Secret_sends_bytearray(self):
 
1633
        ret = [None]
 
1634
        def func(*args, **kwargs):
 
1635
            ret[0] = (args, kwargs)
 
1636
        mock_dbus_python = self.MockDBusPython_func(func)
 
1637
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1638
        bus.set_client_property("objectpath", "Secret", "value")
 
1639
        expected_call = (("se.recompile.Mandos.Client", "Secret",
 
1640
                          mock_dbus_python.ByteArray(b"value")),
 
1641
                         {"dbus_interface":
 
1642
                          "org.freedesktop.DBus.Properties"})
 
1643
        self.assertEqual(expected_call, ret[0])
 
1644
        if sys.version_info.major == 2:
 
1645
            self.assertIsInstance(ret[0][0][-1],
 
1646
                                  mock_dbus_python.ByteArray)
 
1647
 
 
1648
    def test_get_object_converts_to_correct_exception(self):
 
1649
        bus = dbus_python_adapter.SystemBus(
 
1650
            self.fake_dbus_python_raises_exception_on_connect)
 
1651
        with self.assertRaises(dbus.ConnectFailed):
 
1652
            self.call_method(bus, "methodname", "busname",
 
1653
                             "objectpath", "interface")
 
1654
 
 
1655
    class fake_dbus_python_raises_exception_on_connect:
 
1656
        """fake dbus-python module"""
 
1657
        class exceptions:
 
1658
            """Pseudo-namespace"""
 
1659
            class DBusException(Exception):
 
1660
                pass
 
1661
 
 
1662
        @classmethod
 
1663
        def SystemBus(cls):
 
1664
            def get_object(busname, objectpath):
 
1665
                raise cls.exceptions.DBusException()
 
1666
            Bus = collections.namedtuple("Bus", ["get_object"])
 
1667
            return Bus(get_object=get_object)
 
1668
 
 
1669
 
 
1670
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
 
1671
    class mock_dbus_python:
 
1672
        """mock dbus-python modules"""
 
1673
        class SystemBus:
 
1674
            @staticmethod
 
1675
            def get_object(busname, objectpath):
 
1676
                return Unique()
 
1677
 
 
1678
    def setUp(self):
 
1679
        self.bus = dbus_python_adapter.CachingBus(
 
1680
            self.mock_dbus_python)
 
1681
 
 
1682
    def test_returns_distinct_objectpaths(self):
 
1683
        obj1 = self.bus.get_object("busname", "objectpath1")
 
1684
        self.assertIsInstance(obj1, Unique)
 
1685
        obj2 = self.bus.get_object("busname", "objectpath2")
 
1686
        self.assertIsInstance(obj2, Unique)
 
1687
        self.assertIsNot(obj1, obj2)
 
1688
 
 
1689
    def test_returns_distinct_busnames(self):
 
1690
        obj1 = self.bus.get_object("busname1", "objectpath")
 
1691
        self.assertIsInstance(obj1, Unique)
 
1692
        obj2 = self.bus.get_object("busname2", "objectpath")
 
1693
        self.assertIsInstance(obj2, Unique)
 
1694
        self.assertIsNot(obj1, obj2)
 
1695
 
 
1696
    def test_returns_distinct_both(self):
 
1697
        obj1 = self.bus.get_object("busname1", "objectpath")
 
1698
        self.assertIsInstance(obj1, Unique)
 
1699
        obj2 = self.bus.get_object("busname2", "objectpath")
 
1700
        self.assertIsInstance(obj2, Unique)
 
1701
        self.assertIsNot(obj1, obj2)
 
1702
 
 
1703
    def test_returns_same(self):
 
1704
        obj1 = self.bus.get_object("busname", "objectpath")
 
1705
        self.assertIsInstance(obj1, Unique)
 
1706
        obj2 = self.bus.get_object("busname", "objectpath")
 
1707
        self.assertIsInstance(obj2, Unique)
 
1708
        self.assertIs(obj1, obj2)
 
1709
 
 
1710
    def test_returns_same_old(self):
 
1711
        obj1 = self.bus.get_object("busname1", "objectpath1")
 
1712
        self.assertIsInstance(obj1, Unique)
 
1713
        obj2 = self.bus.get_object("busname2", "objectpath2")
 
1714
        self.assertIsInstance(obj2, Unique)
 
1715
        obj1b = self.bus.get_object("busname1", "objectpath1")
 
1716
        self.assertIsInstance(obj1b, Unique)
 
1717
        self.assertIsNot(obj1, obj2)
 
1718
        self.assertIsNot(obj2, obj1b)
 
1719
        self.assertIs(obj1, obj1b)
 
1720
 
 
1721
 
 
1722
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
 
1723
 
 
1724
    def Stub_pydbus_func(self, func):
 
1725
        class stub_pydbus:
 
1726
            """stub pydbus module"""
 
1727
            class SystemBus:
 
1728
                @staticmethod
 
1729
                def get(busname, objectpath):
 
1730
                    DBusObject = collections.namedtuple(
 
1731
                        "DBusObject", ("methodname",))
 
1732
                    return {"interface":
 
1733
                            DBusObject(methodname=func)}
 
1734
        return stub_pydbus
 
1735
 
 
1736
    def call_method(self, bus, methodname, busname, objectpath,
 
1737
                    interface, *args):
 
1738
        with self.assertLogs(log, logging.DEBUG):
 
1739
            return bus.call_method(methodname, busname, objectpath,
 
1740
                                   interface, *args)
 
1741
 
 
1742
    def test_call_method_returns(self):
 
1743
        expected_method_return = Unique()
 
1744
        method_args = (Unique(), Unique())
 
1745
        def func(*args):
 
1746
            self.assertEqual(len(method_args), len(args))
 
1747
            for marg, arg in zip(method_args, args):
 
1748
                self.assertIs(marg, arg)
 
1749
            return expected_method_return
 
1750
        stub_pydbus = self.Stub_pydbus_func(func)
 
1751
        bus = pydbus_adapter.SystemBus(stub_pydbus)
 
1752
        ret = self.call_method(bus, "methodname", "busname",
 
1753
                               "objectpath", "interface",
 
1754
                               *method_args)
 
1755
        self.assertIs(ret, expected_method_return)
 
1756
 
 
1757
    def test_call_method_handles_exception(self):
 
1758
        def func():
 
1759
            raise gi.repository.GLib.Error()
 
1760
 
 
1761
        stub_pydbus = self.Stub_pydbus_func(func)
 
1762
        bus = pydbus_adapter.SystemBus(stub_pydbus)
 
1763
 
 
1764
        with self.assertRaises(dbus.Error) as e:
 
1765
            self.call_method(bus, "methodname", "busname",
 
1766
                             "objectpath", "interface")
 
1767
 
 
1768
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1769
 
 
1770
    def test_get_converts_to_correct_exception(self):
 
1771
        bus = pydbus_adapter.SystemBus(
 
1772
            self.fake_pydbus_raises_exception_on_connect)
 
1773
        with self.assertRaises(dbus.ConnectFailed):
 
1774
            self.call_method(bus, "methodname", "busname",
 
1775
                             "objectpath", "interface")
 
1776
 
 
1777
    class fake_pydbus_raises_exception_on_connect:
 
1778
        """fake dbus-python module"""
 
1779
        @classmethod
 
1780
        def SystemBus(cls):
 
1781
            def get(busname, objectpath):
 
1782
                raise gi.repository.GLib.Error()
 
1783
            Bus = collections.namedtuple("Bus", ["get"])
 
1784
            return Bus(get=get)
 
1785
 
 
1786
    def test_set_property_uses_setattr(self):
 
1787
        class Object:
 
1788
            pass
 
1789
        obj = Object()
 
1790
        class pydbus_spy:
 
1791
            class SystemBus:
 
1792
                @staticmethod
 
1793
                def get(busname, objectpath):
 
1794
                    return {"interface": obj}
 
1795
        bus = pydbus_adapter.SystemBus(pydbus_spy)
 
1796
        value = Unique()
 
1797
        bus.set_property("busname", "objectpath", "interface", "key",
 
1798
                         value)
 
1799
        self.assertIs(value, obj.key)
 
1800
 
 
1801
    def test_get_suppresses_xml_deprecation_warning(self):
 
1802
        if sys.version_info.major >= 3:
 
1803
            return
 
1804
        class stub_pydbus_get:
 
1805
            class SystemBus:
 
1806
                @staticmethod
 
1807
                def get(busname, objectpath):
 
1808
                    warnings.warn_explicit(
 
1809
                        "deprecated", DeprecationWarning,
 
1810
                        "xml.etree.ElementTree", 0)
 
1811
        bus = pydbus_adapter.SystemBus(stub_pydbus_get)
 
1812
        with warnings.catch_warnings(record=True) as w:
 
1813
            warnings.simplefilter("always")
 
1814
            bus.get("busname", "objectpath")
 
1815
            self.assertEqual(0, len(w))
 
1816
 
 
1817
 
 
1818
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
 
1819
    class stub_pydbus:
 
1820
        """stub pydbus module"""
 
1821
        class SystemBus:
 
1822
            @staticmethod
 
1823
            def get(busname, objectpath):
 
1824
                return Unique()
 
1825
 
 
1826
    def setUp(self):
 
1827
        self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
 
1828
 
 
1829
    def test_returns_distinct_objectpaths(self):
 
1830
        obj1 = self.bus.get("busname", "objectpath1")
 
1831
        self.assertIsInstance(obj1, Unique)
 
1832
        obj2 = self.bus.get("busname", "objectpath2")
 
1833
        self.assertIsInstance(obj2, Unique)
 
1834
        self.assertIsNot(obj1, obj2)
 
1835
 
 
1836
    def test_returns_distinct_busnames(self):
 
1837
        obj1 = self.bus.get("busname1", "objectpath")
 
1838
        self.assertIsInstance(obj1, Unique)
 
1839
        obj2 = self.bus.get("busname2", "objectpath")
 
1840
        self.assertIsInstance(obj2, Unique)
 
1841
        self.assertIsNot(obj1, obj2)
 
1842
 
 
1843
    def test_returns_distinct_both(self):
 
1844
        obj1 = self.bus.get("busname1", "objectpath")
 
1845
        self.assertIsInstance(obj1, Unique)
 
1846
        obj2 = self.bus.get("busname2", "objectpath")
 
1847
        self.assertIsInstance(obj2, Unique)
 
1848
        self.assertIsNot(obj1, obj2)
 
1849
 
 
1850
    def test_returns_same(self):
 
1851
        obj1 = self.bus.get("busname", "objectpath")
 
1852
        self.assertIsInstance(obj1, Unique)
 
1853
        obj2 = self.bus.get("busname", "objectpath")
 
1854
        self.assertIsInstance(obj2, Unique)
 
1855
        self.assertIs(obj1, obj2)
 
1856
 
 
1857
    def test_returns_same_old(self):
 
1858
        obj1 = self.bus.get("busname1", "objectpath1")
 
1859
        self.assertIsInstance(obj1, Unique)
 
1860
        obj2 = self.bus.get("busname2", "objectpath2")
 
1861
        self.assertIsInstance(obj2, Unique)
 
1862
        obj1b = self.bus.get("busname1", "objectpath1")
 
1863
        self.assertIsInstance(obj1b, Unique)
 
1864
        self.assertIsNot(obj1, obj2)
 
1865
        self.assertIsNot(obj2, obj1b)
 
1866
        self.assertIs(obj1, obj1b)
 
1867
 
 
1868
 
 
1869
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
 
1870
 
 
1871
    class dummy_dbussy:
 
1872
        class DBUS:
 
1873
            class ObjectPath(str):
 
1874
                pass
 
1875
        class DBusError(Exception):
 
1876
            pass
 
1877
 
 
1878
    def fake_ravel_func(self, func):
 
1879
        class fake_ravel:
 
1880
            @staticmethod
 
1881
            def system_bus():
 
1882
                class DBusInterfaceProxy:
 
1883
                    @staticmethod
 
1884
                    def methodname(*args):
 
1885
                        return [func(*args)]
 
1886
                class DBusObject:
 
1887
                    @staticmethod
 
1888
                    def get_interface(interface):
 
1889
                        if interface == "interface":
 
1890
                            return DBusInterfaceProxy()
 
1891
                return {"busname": {"objectpath": DBusObject()}}
 
1892
        return fake_ravel
 
1893
 
 
1894
    def call_method(self, bus, methodname, busname, objectpath,
 
1895
                    interface, *args):
 
1896
        with self.assertLogs(log, logging.DEBUG):
 
1897
            return bus.call_method(methodname, busname, objectpath,
 
1898
                                   interface, *args)
 
1899
 
 
1900
    def test_call_method_returns(self):
 
1901
        expected_method_return = Unique()
 
1902
        method_args = (Unique(), Unique())
 
1903
        def func(*args):
 
1904
            self.assertEqual(len(method_args), len(args))
 
1905
            for marg, arg in zip(method_args, args):
 
1906
                self.assertIs(marg, arg)
 
1907
            return expected_method_return
 
1908
        fake_ravel = self.fake_ravel_func(func)
 
1909
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1910
        ret = self.call_method(bus, "methodname", "busname",
 
1911
                               "objectpath", "interface",
 
1912
                               *method_args)
 
1913
        self.assertIs(ret, expected_method_return)
 
1914
 
 
1915
    def test_call_method_filters_objectpath(self):
 
1916
        def func():
 
1917
            return method_return
 
1918
        fake_ravel = self.fake_ravel_func(func)
 
1919
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1920
        method_return = (self.dummy_dbussy.DBUS
 
1921
                         .ObjectPath("objectpath"))
 
1922
        ret = self.call_method(bus, "methodname", "busname",
 
1923
                               "objectpath", "interface")
 
1924
        self.assertEqual("objectpath", ret)
 
1925
        self.assertNotIsInstance(ret,
 
1926
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
1927
 
 
1928
    def test_call_method_filters_objectpaths_in_dict(self):
 
1929
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1930
        def func():
 
1931
            return method_return
 
1932
        fake_ravel = self.fake_ravel_func(func)
 
1933
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1934
        method_return = {
 
1935
            ObjectPath("objectpath_key_1"):
 
1936
            ObjectPath("objectpath_value_1"),
 
1937
            ObjectPath("objectpath_key_2"):
 
1938
            ObjectPath("objectpath_value_2"),
 
1939
        }
 
1940
        ret = self.call_method(bus, "methodname", "busname",
 
1941
                               "objectpath", "interface")
 
1942
        expected_method_return = {str(key): str(value)
 
1943
                                  for key, value in
 
1944
                                  method_return.items()}
 
1945
        for key, value in ret.items():
 
1946
            self.assertNotIsInstance(key, ObjectPath)
 
1947
            self.assertNotIsInstance(value, ObjectPath)
 
1948
        self.assertEqual(expected_method_return, ret)
 
1949
        self.assertIsInstance(ret, dict)
 
1950
 
 
1951
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
 
1952
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1953
        def func():
 
1954
            return method_return
 
1955
        fake_ravel = self.fake_ravel_func(func)
 
1956
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1957
        method_return = {
 
1958
            ObjectPath("key1"): {
 
1959
                ObjectPath("key11"): ObjectPath("value11"),
 
1960
                ObjectPath("key12"): ObjectPath("value12"),
 
1961
            },
 
1962
            ObjectPath("key2"): {
 
1963
                ObjectPath("key21"): ObjectPath("value21"),
 
1964
                ObjectPath("key22"): ObjectPath("value22"),
 
1965
            },
 
1966
        }
 
1967
        ret = self.call_method(bus, "methodname", "busname",
 
1968
                               "objectpath", "interface")
 
1969
        expected_method_return = {
 
1970
            "key1": {"key11": "value11",
 
1971
                     "key12": "value12"},
 
1972
            "key2": {"key21": "value21",
 
1973
                     "key22": "value22"},
 
1974
        }
 
1975
        self.assertEqual(expected_method_return, ret)
 
1976
        for key, value in ret.items():
 
1977
            self.assertIsInstance(value, dict)
 
1978
            self.assertEqual(expected_method_return[key], value)
 
1979
            self.assertNotIsInstance(key, ObjectPath)
 
1980
            for inner_key, inner_value in value.items():
 
1981
                self.assertIsInstance(value, dict)
 
1982
                self.assertEqual(
 
1983
                    expected_method_return[key][inner_key],
 
1984
                    inner_value)
 
1985
                self.assertNotIsInstance(key, ObjectPath)
 
1986
 
 
1987
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
 
1988
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1989
        def func():
 
1990
            return method_return
 
1991
        fake_ravel = self.fake_ravel_func(func)
 
1992
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1993
        method_return = {
 
1994
            ObjectPath("key1"): {
 
1995
                ObjectPath("key2"): {
 
1996
                    ObjectPath("key3"): ObjectPath("value"),
 
1997
                },
 
1998
            },
 
1999
        }
 
2000
        ret = self.call_method(bus, "methodname", "busname",
 
2001
                               "objectpath", "interface")
 
2002
        expected_method_return = {"key1": {"key2": {"key3": "value"}}}
 
2003
        self.assertEqual(expected_method_return, ret)
 
2004
        self.assertIsInstance(ret, dict)
 
2005
        self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
 
2006
        self.assertIsInstance(ret["key1"], dict)
 
2007
        self.assertNotIsInstance(next(iter(ret["key1"].keys())),
 
2008
                                 ObjectPath)
 
2009
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
2010
        self.assertNotIsInstance(
 
2011
            next(iter(ret["key1"]["key2"].keys())),
 
2012
            ObjectPath)
 
2013
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
 
2014
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
2015
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
2016
 
 
2017
    def test_call_method_handles_exception(self):
 
2018
        def func():
 
2019
            raise self.dummy_dbussy.DBusError()
 
2020
 
 
2021
        fake_ravel = self.fake_ravel_func(func)
 
2022
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2023
 
 
2024
        with self.assertRaises(dbus.Error) as e:
 
2025
            self.call_method(bus, "methodname", "busname",
 
2026
                             "objectpath", "interface")
 
2027
 
 
2028
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
2029
 
 
2030
    def test_get_object_converts_to_correct_exception(self):
 
2031
        class fake_ravel_raises_exception_on_connect:
 
2032
            @staticmethod
 
2033
            def system_bus():
 
2034
                class Bus:
 
2035
                    @staticmethod
 
2036
                    def __getitem__(key):
 
2037
                        if key == "objectpath":
 
2038
                            raise self.dummy_dbussy.DBusError()
 
2039
                        raise Exception(key)
 
2040
                return {"busname": Bus()}
 
2041
        def func():
 
2042
            raise self.dummy_dbussy.DBusError()
 
2043
        bus = dbussy_adapter.SystemBus(
 
2044
            self.dummy_dbussy,
 
2045
            fake_ravel_raises_exception_on_connect)
 
2046
        with self.assertRaises(dbus.ConnectFailed):
 
2047
            self.call_method(bus, "methodname", "busname",
 
2048
                             "objectpath", "interface")
 
2049
 
 
2050
 
 
2051
class Test_commands_from_options(unittest.TestCase):
 
2052
 
 
2053
    def setUp(self):
 
2054
        self.parser = argparse.ArgumentParser()
 
2055
        add_command_line_options(self.parser)
 
2056
 
 
2057
    def test_is_enabled(self):
 
2058
        self.assert_command_from_args(["--is-enabled", "client"],
 
2059
                                      command.IsEnabled)
 
2060
 
 
2061
    def assert_command_from_args(self, args, command_cls, length=1,
 
2062
                                 clients=None, **cmd_attrs):
 
2063
        """Assert that parsing ARGS should result in an instance of
 
2064
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
 
2065
        options = self.parser.parse_args(args)
 
2066
        check_option_syntax(self.parser, options)
 
2067
        commands = commands_from_options(options)
 
2068
        self.assertEqual(length, len(commands))
 
2069
        for command in commands:
 
2070
            if isinstance(command, command_cls):
 
2071
                break
 
2072
        else:
 
2073
            self.assertIsInstance(command, command_cls)
 
2074
        if clients is not None:
 
2075
            self.assertEqual(clients, options.client)
 
2076
        for key, value in cmd_attrs.items():
 
2077
            self.assertEqual(value, getattr(command, key))
 
2078
 
 
2079
    def assert_commands_from_args(self, args, commands, clients=None):
 
2080
        for cmd in commands:
 
2081
            self.assert_command_from_args(args, cmd,
 
2082
                                          length=len(commands),
 
2083
                                          clients=clients)
 
2084
 
 
2085
    def test_is_enabled_short(self):
 
2086
        self.assert_command_from_args(["-V", "client"],
 
2087
                                      command.IsEnabled)
 
2088
 
 
2089
    def test_approve(self):
 
2090
        self.assert_command_from_args(["--approve", "client"],
 
2091
                                      command.Approve)
 
2092
 
 
2093
    def test_approve_short(self):
 
2094
        self.assert_command_from_args(["-A", "client"],
 
2095
                                      command.Approve)
 
2096
 
 
2097
    def test_deny(self):
 
2098
        self.assert_command_from_args(["--deny", "client"],
 
2099
                                      command.Deny)
 
2100
 
 
2101
    def test_deny_short(self):
 
2102
        self.assert_command_from_args(["-D", "client"], command.Deny)
 
2103
 
 
2104
    def test_remove(self):
 
2105
        self.assert_command_from_args(["--remove", "client"],
 
2106
                                      command.Remove)
 
2107
 
 
2108
    def test_deny_before_remove(self):
 
2109
        options = self.parser.parse_args(["--deny", "--remove",
 
2110
                                          "client"])
 
2111
        check_option_syntax(self.parser, options)
 
2112
        commands = commands_from_options(options)
 
2113
        self.assertEqual(2, len(commands))
 
2114
        self.assertIsInstance(commands[0], command.Deny)
 
2115
        self.assertIsInstance(commands[1], command.Remove)
 
2116
 
 
2117
    def test_deny_before_remove_reversed(self):
 
2118
        options = self.parser.parse_args(["--remove", "--deny",
 
2119
                                          "--all"])
 
2120
        check_option_syntax(self.parser, options)
 
2121
        commands = commands_from_options(options)
 
2122
        self.assertEqual(2, len(commands))
 
2123
        self.assertIsInstance(commands[0], command.Deny)
 
2124
        self.assertIsInstance(commands[1], command.Remove)
 
2125
 
 
2126
    def test_remove_short(self):
 
2127
        self.assert_command_from_args(["-r", "client"],
 
2128
                                      command.Remove)
 
2129
 
 
2130
    def test_dump_json(self):
 
2131
        self.assert_command_from_args(["--dump-json"],
 
2132
                                      command.DumpJSON)
 
2133
 
 
2134
    def test_enable(self):
 
2135
        self.assert_command_from_args(["--enable", "client"],
 
2136
                                      command.Enable)
 
2137
 
 
2138
    def test_enable_short(self):
 
2139
        self.assert_command_from_args(["-e", "client"],
 
2140
                                      command.Enable)
 
2141
 
 
2142
    def test_disable(self):
 
2143
        self.assert_command_from_args(["--disable", "client"],
 
2144
                                      command.Disable)
 
2145
 
 
2146
    def test_disable_short(self):
 
2147
        self.assert_command_from_args(["-d", "client"],
 
2148
                                      command.Disable)
 
2149
 
 
2150
    def test_bump_timeout(self):
 
2151
        self.assert_command_from_args(["--bump-timeout", "client"],
 
2152
                                      command.BumpTimeout)
 
2153
 
 
2154
    def test_bump_timeout_short(self):
 
2155
        self.assert_command_from_args(["-b", "client"],
 
2156
                                      command.BumpTimeout)
 
2157
 
 
2158
    def test_start_checker(self):
 
2159
        self.assert_command_from_args(["--start-checker", "client"],
 
2160
                                      command.StartChecker)
 
2161
 
 
2162
    def test_stop_checker(self):
 
2163
        self.assert_command_from_args(["--stop-checker", "client"],
 
2164
                                      command.StopChecker)
 
2165
 
 
2166
    def test_approve_by_default(self):
 
2167
        self.assert_command_from_args(["--approve-by-default",
 
2168
                                       "client"],
 
2169
                                      command.ApproveByDefault)
 
2170
 
 
2171
    def test_deny_by_default(self):
 
2172
        self.assert_command_from_args(["--deny-by-default", "client"],
 
2173
                                      command.DenyByDefault)
 
2174
 
 
2175
    def test_checker(self):
 
2176
        self.assert_command_from_args(["--checker", ":", "client"],
 
2177
                                      command.SetChecker,
 
2178
                                      value_to_set=":")
 
2179
 
 
2180
    def test_checker_empty(self):
 
2181
        self.assert_command_from_args(["--checker", "", "client"],
 
2182
                                      command.SetChecker,
 
2183
                                      value_to_set="")
 
2184
 
 
2185
    def test_checker_short(self):
 
2186
        self.assert_command_from_args(["-c", ":", "client"],
 
2187
                                      command.SetChecker,
 
2188
                                      value_to_set=":")
 
2189
 
 
2190
    def test_host(self):
 
2191
        self.assert_command_from_args(
 
2192
            ["--host", "client.example.org", "client"],
 
2193
            command.SetHost, value_to_set="client.example.org")
 
2194
 
 
2195
    def test_host_short(self):
 
2196
        self.assert_command_from_args(
 
2197
            ["-H", "client.example.org", "client"], command.SetHost,
 
2198
            value_to_set="client.example.org")
 
2199
 
 
2200
    def test_secret_devnull(self):
 
2201
        self.assert_command_from_args(["--secret", os.path.devnull,
 
2202
                                       "client"], command.SetSecret,
 
2203
                                      value_to_set=b"")
 
2204
 
 
2205
    def test_secret_tempfile(self):
 
2206
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
 
2207
            value = b"secret\0xyzzy\nbar"
 
2208
            f.write(value)
 
2209
            f.seek(0)
 
2210
            self.assert_command_from_args(["--secret", f.name,
 
2211
                                           "client"],
 
2212
                                          command.SetSecret,
 
2213
                                          value_to_set=value)
 
2214
 
 
2215
    def test_secret_devnull_short(self):
 
2216
        self.assert_command_from_args(["-s", os.path.devnull,
 
2217
                                       "client"], command.SetSecret,
 
2218
                                      value_to_set=b"")
 
2219
 
 
2220
    def test_secret_tempfile_short(self):
 
2221
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
 
2222
            value = b"secret\0xyzzy\nbar"
 
2223
            f.write(value)
 
2224
            f.seek(0)
 
2225
            self.assert_command_from_args(["-s", f.name, "client"],
 
2226
                                          command.SetSecret,
 
2227
                                          value_to_set=value)
 
2228
 
 
2229
    def test_timeout(self):
 
2230
        self.assert_command_from_args(["--timeout", "PT5M", "client"],
 
2231
                                      command.SetTimeout,
 
2232
                                      value_to_set=300000)
 
2233
 
 
2234
    def test_timeout_short(self):
 
2235
        self.assert_command_from_args(["-t", "PT5M", "client"],
 
2236
                                      command.SetTimeout,
 
2237
                                      value_to_set=300000)
 
2238
 
 
2239
    def test_extended_timeout(self):
 
2240
        self.assert_command_from_args(["--extended-timeout", "PT15M",
 
2241
                                       "client"],
 
2242
                                      command.SetExtendedTimeout,
 
2243
                                      value_to_set=900000)
 
2244
 
 
2245
    def test_interval(self):
 
2246
        self.assert_command_from_args(["--interval", "PT2M",
 
2247
                                       "client"], command.SetInterval,
 
2248
                                      value_to_set=120000)
 
2249
 
 
2250
    def test_interval_short(self):
 
2251
        self.assert_command_from_args(["-i", "PT2M", "client"],
 
2252
                                      command.SetInterval,
 
2253
                                      value_to_set=120000)
 
2254
 
 
2255
    def test_approval_delay(self):
 
2256
        self.assert_command_from_args(["--approval-delay", "PT30S",
 
2257
                                       "client"],
 
2258
                                      command.SetApprovalDelay,
 
2259
                                      value_to_set=30000)
 
2260
 
 
2261
    def test_approval_duration(self):
 
2262
        self.assert_command_from_args(["--approval-duration", "PT1S",
 
2263
                                       "client"],
 
2264
                                      command.SetApprovalDuration,
 
2265
                                      value_to_set=1000)
 
2266
 
 
2267
    def test_print_table(self):
 
2268
        self.assert_command_from_args([], command.PrintTable,
 
2269
                                      verbose=False)
 
2270
 
 
2271
    def test_print_table_verbose(self):
 
2272
        self.assert_command_from_args(["--verbose"],
 
2273
                                      command.PrintTable,
 
2274
                                      verbose=True)
 
2275
 
 
2276
    def test_print_table_verbose_short(self):
 
2277
        self.assert_command_from_args(["-v"], command.PrintTable,
 
2278
                                      verbose=True)
 
2279
 
 
2280
 
 
2281
    def test_manual_page_example_1(self):
 
2282
        self.assert_command_from_args("",
 
2283
                                      command.PrintTable,
 
2284
                                      clients=[],
 
2285
                                      verbose=False)
 
2286
 
 
2287
    def test_manual_page_example_2(self):
 
2288
        self.assert_command_from_args(
 
2289
            "--verbose foo1.example.org foo2.example.org".split(),
 
2290
            command.PrintTable, clients=["foo1.example.org",
 
2291
                                         "foo2.example.org"],
 
2292
            verbose=True)
 
2293
 
 
2294
    def test_manual_page_example_3(self):
 
2295
        self.assert_command_from_args("--enable --all".split(),
 
2296
                                      command.Enable,
 
2297
                                      clients=[])
 
2298
 
 
2299
    def test_manual_page_example_4(self):
 
2300
        self.assert_commands_from_args(
 
2301
            ("--timeout=PT5M --interval=PT1M foo1.example.org"
 
2302
             " foo2.example.org").split(),
 
2303
            [command.SetTimeout, command.SetInterval],
 
2304
            clients=["foo1.example.org", "foo2.example.org"])
 
2305
 
 
2306
    def test_manual_page_example_5(self):
 
2307
        self.assert_command_from_args("--approve --all".split(),
 
2308
                                      command.Approve,
 
2309
                                      clients=[])
 
2310
 
 
2311
 
 
2312
class TestCommand(unittest.TestCase):
 
2313
    """Abstract class for tests of command classes"""
 
2314
 
 
2315
    class FakeMandosBus(dbus.MandosBus):
 
2316
        def __init__(self, testcase):
 
2317
            self.client_properties = {
 
2318
                "Name": "foo",
 
2319
                "KeyID": ("92ed150794387c03ce684574b1139a65"
 
2320
                          "94a34f895daaaf09fd8ea90a27cddb12"),
 
2321
                "Secret": b"secret",
 
2322
                "Host": "foo.example.org",
 
2323
                "Enabled": True,
 
2324
                "Timeout": 300000,
 
2325
                "LastCheckedOK": "2019-02-03T00:00:00",
 
2326
                "Created": "2019-01-02T00:00:00",
 
2327
                "Interval": 120000,
 
2328
                "Fingerprint": ("778827225BA7DE539C5A"
 
2329
                                "7CFA59CFF7CDBD9A5920"),
 
2330
                "CheckerRunning": False,
 
2331
                "LastEnabled": "2019-01-03T00:00:00",
 
2332
                "ApprovalPending": False,
 
2333
                "ApprovedByDefault": True,
 
2334
                "LastApprovalRequest": "",
 
2335
                "ApprovalDelay": 0,
 
2336
                "ApprovalDuration": 1000,
 
2337
                "Checker": "fping -q -- %(host)s",
 
2338
                "ExtendedTimeout": 900000,
 
2339
                "Expires": "2019-02-04T00:00:00",
 
2340
                "LastCheckerStatus": 0,
 
2341
            }
 
2342
            self.other_client_properties = {
 
2343
                "Name": "barbar",
 
2344
                "KeyID": ("0558568eedd67d622f5c83b35a115f79"
 
2345
                          "6ab612cff5ad227247e46c2b020f441c"),
 
2346
                "Secret": b"secretbar",
 
2347
                "Host": "192.0.2.3",
 
2348
                "Enabled": True,
 
2349
                "Timeout": 300000,
 
2350
                "LastCheckedOK": "2019-02-04T00:00:00",
 
2351
                "Created": "2019-01-03T00:00:00",
 
2352
                "Interval": 120000,
 
2353
                "Fingerprint": ("3E393AEAEFB84C7E89E2"
 
2354
                                "F547B3A107558FCA3A27"),
 
2355
                "CheckerRunning": True,
 
2356
                "LastEnabled": "2019-01-04T00:00:00",
 
2357
                "ApprovalPending": False,
 
2358
                "ApprovedByDefault": False,
 
2359
                "LastApprovalRequest": "2019-01-03T00:00:00",
 
2360
                "ApprovalDelay": 30000,
 
2361
                "ApprovalDuration": 93785000,
 
2362
                "Checker": ":",
 
2363
                "ExtendedTimeout": 900000,
 
2364
                "Expires": "2019-02-05T00:00:00",
 
2365
                "LastCheckerStatus": -2,
 
2366
            }
 
2367
            self.clients =  collections.OrderedDict(
 
2368
                [
 
2369
                    ("client_objectpath", self.client_properties),
 
2370
                    ("other_client_objectpath",
 
2371
                     self.other_client_properties),
 
2372
                ])
 
2373
            self.one_client = {"client_objectpath":
 
2374
                               self.client_properties}
 
2375
            self.testcase = testcase
 
2376
            self.calls = []
 
2377
 
 
2378
        def call_method(self, methodname, busname, objectpath,
 
2379
                        interface, *args):
 
2380
            self.testcase.assertEqual("se.recompile.Mandos", busname)
 
2381
            self.calls.append((methodname, busname, objectpath,
 
2382
                               interface, args))
 
2383
            if interface == "org.freedesktop.DBus.Properties":
 
2384
                if methodname == "Set":
 
2385
                    self.testcase.assertEqual(3, len(args))
 
2386
                    interface, key, value = args
 
2387
                    self.testcase.assertEqual(
 
2388
                        "se.recompile.Mandos.Client", interface)
 
2389
                    self.clients[objectpath][key] = value
 
2390
                    return
 
2391
            elif interface == "se.recompile.Mandos":
 
2392
                self.testcase.assertEqual("RemoveClient", methodname)
 
2393
                self.testcase.assertEqual(1, len(args))
 
2394
                clientpath = args[0]
 
2395
                del self.clients[clientpath]
 
2396
                return
 
2397
            elif interface == "se.recompile.Mandos.Client":
 
2398
                if methodname == "Approve":
 
2399
                    self.testcase.assertEqual(1, len(args))
 
2400
                    return
 
2401
            raise ValueError()
 
2402
 
 
2403
    def setUp(self):
 
2404
        self.bus = self.FakeMandosBus(self)
 
2405
 
 
2406
 
 
2407
class TestBaseCommands(TestCommand):
 
2408
 
 
2409
    def test_IsEnabled_exits_successfully(self):
 
2410
        with self.assertRaises(SystemExit) as e:
 
2411
            command.IsEnabled().run(self.bus.one_client)
 
2412
        if e.exception.code is not None:
 
2413
            self.assertEqual(0, e.exception.code)
 
2414
        else:
 
2415
            self.assertIsNone(e.exception.code)
 
2416
 
 
2417
    def test_IsEnabled_exits_with_failure(self):
 
2418
        self.bus.client_properties["Enabled"] = False
 
2419
        with self.assertRaises(SystemExit) as e:
 
2420
            command.IsEnabled().run(self.bus.one_client)
 
2421
        if isinstance(e.exception.code, int):
 
2422
            self.assertNotEqual(0, e.exception.code)
 
2423
        else:
 
2424
            self.assertIsNotNone(e.exception.code)
 
2425
 
 
2426
    def test_Approve(self):
 
2427
        busname = "se.recompile.Mandos"
 
2428
        client_interface = "se.recompile.Mandos.Client"
 
2429
        command.Approve().run(self.bus.clients, self.bus)
 
2430
        self.assertTrue(self.bus.clients)
 
2431
        for clientpath in self.bus.clients:
 
2432
            self.assertIn(("Approve", busname, clientpath,
 
2433
                           client_interface, (True,)), self.bus.calls)
 
2434
 
 
2435
    def test_Deny(self):
 
2436
        busname = "se.recompile.Mandos"
 
2437
        client_interface = "se.recompile.Mandos.Client"
 
2438
        command.Deny().run(self.bus.clients, self.bus)
 
2439
        self.assertTrue(self.bus.clients)
 
2440
        for clientpath in self.bus.clients:
 
2441
            self.assertIn(("Approve", busname, clientpath,
 
2442
                           client_interface, (False,)),
 
2443
                          self.bus.calls)
 
2444
 
 
2445
    def test_Remove(self):
 
2446
        busname = "se.recompile.Mandos"
 
2447
        server_path = "/"
 
2448
        server_interface = "se.recompile.Mandos"
 
2449
        orig_clients = self.bus.clients.copy()
 
2450
        command.Remove().run(self.bus.clients, self.bus)
 
2451
        self.assertFalse(self.bus.clients)
 
2452
        for clientpath in orig_clients:
 
2453
            self.assertIn(("RemoveClient", busname,
 
2454
                           server_path, server_interface,
 
2455
                           (clientpath,)), self.bus.calls)
 
2456
 
 
2457
    expected_json = {
 
2458
        "foo": {
 
2459
            "Name": "foo",
 
2460
            "KeyID": ("92ed150794387c03ce684574b1139a65"
 
2461
                      "94a34f895daaaf09fd8ea90a27cddb12"),
 
2462
            "Host": "foo.example.org",
 
2463
            "Enabled": True,
 
2464
            "Timeout": 300000,
 
2465
            "LastCheckedOK": "2019-02-03T00:00:00",
 
2466
            "Created": "2019-01-02T00:00:00",
 
2467
            "Interval": 120000,
 
2468
            "Fingerprint": ("778827225BA7DE539C5A"
 
2469
                            "7CFA59CFF7CDBD9A5920"),
 
2470
            "CheckerRunning": False,
 
2471
            "LastEnabled": "2019-01-03T00:00:00",
 
2472
            "ApprovalPending": False,
 
2473
            "ApprovedByDefault": True,
 
2474
            "LastApprovalRequest": "",
 
2475
            "ApprovalDelay": 0,
 
2476
            "ApprovalDuration": 1000,
 
2477
            "Checker": "fping -q -- %(host)s",
 
2478
            "ExtendedTimeout": 900000,
 
2479
            "Expires": "2019-02-04T00:00:00",
 
2480
            "LastCheckerStatus": 0,
 
2481
        },
 
2482
        "barbar": {
 
2483
            "Name": "barbar",
 
2484
            "KeyID": ("0558568eedd67d622f5c83b35a115f79"
 
2485
                      "6ab612cff5ad227247e46c2b020f441c"),
 
2486
            "Host": "192.0.2.3",
 
2487
            "Enabled": True,
 
2488
            "Timeout": 300000,
 
2489
            "LastCheckedOK": "2019-02-04T00:00:00",
 
2490
            "Created": "2019-01-03T00:00:00",
 
2491
            "Interval": 120000,
 
2492
            "Fingerprint": ("3E393AEAEFB84C7E89E2"
 
2493
                            "F547B3A107558FCA3A27"),
 
2494
            "CheckerRunning": True,
 
2495
            "LastEnabled": "2019-01-04T00:00:00",
 
2496
            "ApprovalPending": False,
 
2497
            "ApprovedByDefault": False,
 
2498
            "LastApprovalRequest": "2019-01-03T00:00:00",
 
2499
            "ApprovalDelay": 30000,
 
2500
            "ApprovalDuration": 93785000,
 
2501
            "Checker": ":",
 
2502
            "ExtendedTimeout": 900000,
 
2503
            "Expires": "2019-02-05T00:00:00",
 
2504
            "LastCheckerStatus": -2,
 
2505
        },
 
2506
    }
 
2507
 
 
2508
    def test_DumpJSON_normal(self):
 
2509
        with self.capture_stdout_to_buffer() as buffer:
 
2510
            command.DumpJSON().run(self.bus.clients)
 
2511
        json_data = json.loads(buffer.getvalue())
 
2512
        self.assertDictEqual(self.expected_json, json_data)
 
2513
 
 
2514
    @staticmethod
 
2515
    @contextlib.contextmanager
 
2516
    def capture_stdout_to_buffer():
 
2517
        capture_buffer = io.StringIO()
 
2518
        old_stdout = sys.stdout
 
2519
        sys.stdout = capture_buffer
 
2520
        try:
 
2521
            yield capture_buffer
 
2522
        finally:
 
2523
            sys.stdout = old_stdout
 
2524
 
 
2525
    def test_DumpJSON_one_client(self):
 
2526
        with self.capture_stdout_to_buffer() as buffer:
 
2527
            command.DumpJSON().run(self.bus.one_client)
 
2528
        json_data = json.loads(buffer.getvalue())
 
2529
        expected_json = {"foo": self.expected_json["foo"]}
 
2530
        self.assertDictEqual(expected_json, json_data)
 
2531
 
 
2532
    def test_PrintTable_normal(self):
 
2533
        with self.capture_stdout_to_buffer() as buffer:
 
2534
            command.PrintTable().run(self.bus.clients)
 
2535
        expected_output = "\n".join((
 
2536
            "Name   Enabled Timeout  Last Successful Check",
 
2537
            "foo    Yes     00:05:00 2019-02-03T00:00:00  ",
 
2538
            "barbar Yes     00:05:00 2019-02-04T00:00:00  ",
 
2539
        )) + "\n"
 
2540
        self.assertEqual(expected_output, buffer.getvalue())
 
2541
 
 
2542
    def test_PrintTable_verbose(self):
 
2543
        with self.capture_stdout_to_buffer() as buffer:
 
2544
            command.PrintTable(verbose=True).run(self.bus.clients)
 
2545
        columns = (
 
2546
            (
 
2547
                "Name   ",
 
2548
                "foo    ",
 
2549
                "barbar ",
 
2550
            ),(
 
2551
                "Enabled ",
 
2552
                "Yes     ",
 
2553
                "Yes     ",
 
2554
            ),(
 
2555
                "Timeout  ",
 
2556
                "00:05:00 ",
 
2557
                "00:05:00 ",
 
2558
            ),(
 
2559
                "Last Successful Check ",
 
2560
                "2019-02-03T00:00:00   ",
 
2561
                "2019-02-04T00:00:00   ",
 
2562
            ),(
 
2563
                "Created             ",
 
2564
                "2019-01-02T00:00:00 ",
 
2565
                "2019-01-03T00:00:00 ",
 
2566
            ),(
 
2567
                "Interval ",
 
2568
                "00:02:00 ",
 
2569
                "00:02:00 ",
 
2570
            ),(
 
2571
                "Host            ",
 
2572
                "foo.example.org ",
 
2573
                "192.0.2.3       ",
 
2574
            ),(
 
2575
                ("Key ID                                             "
 
2576
                 "              "),
 
2577
                ("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
 
2578
                 "ea90a27cddb12 "),
 
2579
                ("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
 
2580
                 "46c2b020f441c "),
 
2581
            ),(
 
2582
                "Fingerprint                              ",
 
2583
                "778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
 
2584
                "3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
 
2585
            ),(
 
2586
                "Check Is Running ",
 
2587
                "No               ",
 
2588
                "Yes              ",
 
2589
            ),(
 
2590
                "Last Enabled        ",
 
2591
                "2019-01-03T00:00:00 ",
 
2592
                "2019-01-04T00:00:00 ",
 
2593
            ),(
 
2594
                "Approval Is Pending ",
 
2595
                "No                  ",
 
2596
                "No                  ",
 
2597
            ),(
 
2598
                "Approved By Default ",
 
2599
                "Yes                 ",
 
2600
                "No                  ",
 
2601
            ),(
 
2602
                "Last Approval Request ",
 
2603
                "                      ",
 
2604
                "2019-01-03T00:00:00   ",
 
2605
            ),(
 
2606
                "Approval Delay ",
 
2607
                "00:00:00       ",
 
2608
                "00:00:30       ",
 
2609
            ),(
 
2610
                "Approval Duration ",
 
2611
                "00:00:01          ",
 
2612
                "1T02:03:05        ",
 
2613
            ),(
 
2614
                "Checker              ",
 
2615
                "fping -q -- %(host)s ",
 
2616
                ":                    ",
 
2617
            ),(
 
2618
                "Extended Timeout ",
 
2619
                "00:15:00         ",
 
2620
                "00:15:00         ",
 
2621
            ),(
 
2622
                "Expires             ",
 
2623
                "2019-02-04T00:00:00 ",
 
2624
                "2019-02-05T00:00:00 ",
 
2625
            ),(
 
2626
                "Last Checker Status",
 
2627
                "0                  ",
 
2628
                "-2                 ",
 
2629
            )
 
2630
        )
 
2631
        num_lines = max(len(rows) for rows in columns)
 
2632
        expected_output = ("\n".join("".join(rows[line]
 
2633
                                             for rows in columns)
 
2634
                                     for line in range(num_lines))
 
2635
                           + "\n")
 
2636
        self.assertEqual(expected_output, buffer.getvalue())
 
2637
 
 
2638
    def test_PrintTable_one_client(self):
 
2639
        with self.capture_stdout_to_buffer() as buffer:
 
2640
            command.PrintTable().run(self.bus.one_client)
 
2641
        expected_output = "\n".join((
 
2642
            "Name Enabled Timeout  Last Successful Check",
 
2643
            "foo  Yes     00:05:00 2019-02-03T00:00:00  ",
 
2644
        )) + "\n"
 
2645
        self.assertEqual(expected_output, buffer.getvalue())
 
2646
 
 
2647
 
 
2648
class TestPropertySetterCmd(TestCommand):
 
2649
    """Abstract class for tests of command.PropertySetter classes"""
 
2650
 
 
2651
    def runTest(self):
 
2652
        if not hasattr(self, "command"):
 
2653
            return              # Abstract TestCase class
 
2654
 
 
2655
        if hasattr(self, "values_to_set"):
 
2656
            cmd_args = [(value,) for value in self.values_to_set]
 
2657
            values_to_get = getattr(self, "values_to_get",
 
2658
                                    self.values_to_set)
 
2659
        else:
 
2660
            cmd_args = [() for x in range(len(self.values_to_get))]
 
2661
            values_to_get = self.values_to_get
 
2662
        self.assertTrue(values_to_get)
 
2663
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
 
2664
            for clientpath in self.bus.clients:
 
2665
                self.bus.clients[clientpath][self.propname] = (
 
2666
                    Unique())
 
2667
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
 
2668
            self.assertTrue(self.bus.clients)
 
2669
            for clientpath in self.bus.clients:
 
2670
                value = (self.bus.clients[clientpath]
 
2671
                         [self.propname])
 
2672
                self.assertNotIsInstance(value, Unique)
 
2673
                self.assertEqual(value_to_get, value)
 
2674
 
 
2675
 
 
2676
class TestEnableCmd(TestPropertySetterCmd):
 
2677
    command = command.Enable
 
2678
    propname = "Enabled"
 
2679
    values_to_get = [True]
 
2680
 
 
2681
 
 
2682
class TestDisableCmd(TestPropertySetterCmd):
 
2683
    command = command.Disable
 
2684
    propname = "Enabled"
 
2685
    values_to_get = [False]
 
2686
 
 
2687
 
 
2688
class TestBumpTimeoutCmd(TestPropertySetterCmd):
 
2689
    command = command.BumpTimeout
 
2690
    propname = "LastCheckedOK"
 
2691
    values_to_get = [""]
 
2692
 
 
2693
 
 
2694
class TestStartCheckerCmd(TestPropertySetterCmd):
 
2695
    command = command.StartChecker
 
2696
    propname = "CheckerRunning"
 
2697
    values_to_get = [True]
 
2698
 
 
2699
 
 
2700
class TestStopCheckerCmd(TestPropertySetterCmd):
 
2701
    command = command.StopChecker
 
2702
    propname = "CheckerRunning"
 
2703
    values_to_get = [False]
 
2704
 
 
2705
 
 
2706
class TestApproveByDefaultCmd(TestPropertySetterCmd):
 
2707
    command = command.ApproveByDefault
 
2708
    propname = "ApprovedByDefault"
 
2709
    values_to_get = [True]
 
2710
 
 
2711
 
 
2712
class TestDenyByDefaultCmd(TestPropertySetterCmd):
 
2713
    command = command.DenyByDefault
 
2714
    propname = "ApprovedByDefault"
 
2715
    values_to_get = [False]
 
2716
 
 
2717
 
 
2718
class TestSetCheckerCmd(TestPropertySetterCmd):
 
2719
    command = command.SetChecker
 
2720
    propname = "Checker"
 
2721
    values_to_set = ["", ":", "fping -q -- %s"]
 
2722
 
 
2723
 
 
2724
class TestSetHostCmd(TestPropertySetterCmd):
 
2725
    command = command.SetHost
 
2726
    propname = "Host"
 
2727
    values_to_set = ["192.0.2.3", "client.example.org"]
 
2728
 
 
2729
 
 
2730
class TestSetSecretCmd(TestPropertySetterCmd):
 
2731
    command = command.SetSecret
 
2732
    propname = "Secret"
 
2733
    def __init__(self, *args, **kwargs):
 
2734
        self.values_to_set = [io.BytesIO(b""),
 
2735
                              io.BytesIO(b"secret\0xyzzy\nbar")]
 
2736
        self.values_to_get = [f.getvalue() for f in
 
2737
                              self.values_to_set]
 
2738
        super(TestSetSecretCmd, self).__init__(*args, **kwargs)
 
2739
 
 
2740
 
 
2741
class TestSetTimeoutCmd(TestPropertySetterCmd):
 
2742
    command = command.SetTimeout
 
2743
    propname = "Timeout"
 
2744
    values_to_set = [datetime.timedelta(),
 
2745
                     datetime.timedelta(minutes=5),
 
2746
                     datetime.timedelta(seconds=1),
 
2747
                     datetime.timedelta(weeks=1),
 
2748
                     datetime.timedelta(weeks=52)]
 
2749
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2750
 
 
2751
 
 
2752
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
 
2753
    command = command.SetExtendedTimeout
 
2754
    propname = "ExtendedTimeout"
 
2755
    values_to_set = [datetime.timedelta(),
 
2756
                     datetime.timedelta(minutes=5),
 
2757
                     datetime.timedelta(seconds=1),
 
2758
                     datetime.timedelta(weeks=1),
 
2759
                     datetime.timedelta(weeks=52)]
 
2760
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2761
 
 
2762
 
 
2763
class TestSetIntervalCmd(TestPropertySetterCmd):
 
2764
    command = command.SetInterval
 
2765
    propname = "Interval"
 
2766
    values_to_set = [datetime.timedelta(),
 
2767
                     datetime.timedelta(minutes=5),
 
2768
                     datetime.timedelta(seconds=1),
 
2769
                     datetime.timedelta(weeks=1),
 
2770
                     datetime.timedelta(weeks=52)]
 
2771
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2772
 
 
2773
 
 
2774
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
 
2775
    command = command.SetApprovalDelay
 
2776
    propname = "ApprovalDelay"
 
2777
    values_to_set = [datetime.timedelta(),
 
2778
                     datetime.timedelta(minutes=5),
 
2779
                     datetime.timedelta(seconds=1),
 
2780
                     datetime.timedelta(weeks=1),
 
2781
                     datetime.timedelta(weeks=52)]
 
2782
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2783
 
 
2784
 
 
2785
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
 
2786
    command = command.SetApprovalDuration
 
2787
    propname = "ApprovalDuration"
 
2788
    values_to_set = [datetime.timedelta(),
 
2789
                     datetime.timedelta(minutes=5),
 
2790
                     datetime.timedelta(seconds=1),
 
2791
                     datetime.timedelta(weeks=1),
 
2792
                     datetime.timedelta(weeks=52)]
 
2793
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
2794
 
 
2795
 
 
2796
 
 
2797
def parse_test_args():
 
2798
    # type: () -> argparse.Namespace
 
2799
    parser = argparse.ArgumentParser(add_help=False)
 
2800
    parser.add_argument("--check", action="store_true")
 
2801
    parser.add_argument("--prefix", )
 
2802
    args, unknown_args = parser.parse_known_args()
 
2803
    if args.check:
 
2804
        # Remove test options from sys.argv
 
2805
        sys.argv[1:] = unknown_args
 
2806
    return args
 
2807
 
 
2808
# Add all tests from doctest strings
 
2809
def load_tests(loader, tests, none):
 
2810
    import doctest
 
2811
    tests.addTests(doctest.DocTestSuite())
 
2812
    return tests
354
2813
 
355
2814
if __name__ == "__main__":
356
 
    main()
 
2815
    options = parse_test_args()
 
2816
    try:
 
2817
        if options.check:
 
2818
            extra_test_prefix = options.prefix
 
2819
            if extra_test_prefix is not None:
 
2820
                if not (unittest.main(argv=[""], exit=False)
 
2821
                        .result.wasSuccessful()):
 
2822
                    sys.exit(1)
 
2823
                class ExtraTestLoader(unittest.TestLoader):
 
2824
                    testMethodPrefix = extra_test_prefix
 
2825
                # Call using ./scriptname --check [--verbose]
 
2826
                unittest.main(argv=[""], testLoader=ExtraTestLoader())
 
2827
            else:
 
2828
                unittest.main(argv=[""])
 
2829
        else:
 
2830
            main()
 
2831
    finally:
 
2832
        logging.shutdown()
 
2833
 
 
2834
# Local Variables:
 
2835
# run-tests:
 
2836
# (lambda (&optional extra)
 
2837
#   (if (not (funcall run-tests-in-test-buffer default-directory
 
2838
#             extra))
 
2839
#       (funcall show-test-buffer-in-test-window)
 
2840
#     (funcall remove-test-window)
 
2841
#     (if extra (message "Extra tests run successfully!"))))
 
2842
# run-tests-in-test-buffer:
 
2843
# (lambda (dir &optional extra)
 
2844
#   (with-current-buffer (get-buffer-create "*Test*")
 
2845
#     (setq buffer-read-only nil
 
2846
#           default-directory dir)
 
2847
#     (erase-buffer)
 
2848
#     (compilation-mode))
 
2849
#   (let ((process-result
 
2850
#          (let ((inhibit-read-only t))
 
2851
#            (process-file-shell-command
 
2852
#             (funcall get-command-line extra) nil "*Test*"))))
 
2853
#     (and (numberp process-result)
 
2854
#          (= process-result 0))))
 
2855
# get-command-line:
 
2856
# (lambda (&optional extra)
 
2857
#   (let ((quoted-script
 
2858
#          (shell-quote-argument (funcall get-script-name))))
 
2859
#     (format
 
2860
#      (concat "%s --check" (if extra " --prefix=atest" ""))
 
2861
#      quoted-script)))
 
2862
# get-script-name:
 
2863
# (lambda ()
 
2864
#   (if (fboundp 'file-local-name)
 
2865
#       (file-local-name (buffer-file-name))
 
2866
#     (or (file-remote-p (buffer-file-name) 'localname)
 
2867
#         (buffer-file-name))))
 
2868
# remove-test-window:
 
2869
# (lambda ()
 
2870
#   (let ((test-window (get-buffer-window "*Test*")))
 
2871
#     (if test-window (delete-window test-window))))
 
2872
# show-test-buffer-in-test-window:
 
2873
# (lambda ()
 
2874
#   (when (not (get-buffer-window-list "*Test*"))
 
2875
#     (setq next-error-last-buffer (get-buffer "*Test*"))
 
2876
#     (let* ((side (if (>= (window-width) 146) 'right 'bottom))
 
2877
#            (display-buffer-overriding-action
 
2878
#             `((display-buffer-in-side-window) (side . ,side)
 
2879
#               (window-height . fit-window-to-buffer)
 
2880
#               (window-width . fit-window-to-buffer))))
 
2881
#       (display-buffer "*Test*"))))
 
2882
# eval:
 
2883
# (progn
 
2884
#   (let* ((run-extra-tests (lambda () (interactive)
 
2885
#                             (funcall run-tests t)))
 
2886
#          (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
 
2887
#          (outer-keymap `(keymap (3 . ,inner-keymap))))     ; C-c
 
2888
#     (setq minor-mode-overriding-map-alist
 
2889
#           (cons `(run-tests . ,outer-keymap)
 
2890
#                 minor-mode-overriding-map-alist)))
 
2891
#   (add-hook 'after-save-hook run-tests 90 t))
 
2892
# End: