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