/mandos/release

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

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-12 17:33:44 UTC
  • mto: This revision was merged to the branch mainline in revision 382.
  • Revision ID: teddy@recompile.se-20190312173344-13d0hhhkmwu3up78
mandos-ctl: Refactor

* mandos-ctl (domain): Rename to "dbus_busname_domain".  Delete after
                       use.
  (busname): Rename to "dbus_busname".  All users changed.
  (server_path): Rename to "server_dbus_path".  All users changed.
  (server_interface): Rename to "server_dbus_interface".  All users
                      changed.
  (client_interface): Rename to "client_dbus_interface".  All users
                      changed.

Show diffs side-by-side

added added

removed removed

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