2
# -*- mode: python; coding: utf-8 -*-
2
# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-
4
4
# Mandos Monitor - Control and monitor the Mandos server
6
# Copyright © 2008-2010 Teddy Hogeborn
7
# Copyright © 2008-2010 Björn Påhlsson
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
6
# Copyright © 2008-2019 Teddy Hogeborn
7
# Copyright © 2008-2019 Björn Påhlsson
9
# This file is part of Mandos.
11
# Mandos is free software: you can redistribute it and/or modify it
12
# under the terms of the GNU General Public License as published by
11
13
# the Free Software Foundation, either version 3 of the License, or
12
14
# (at your option) any later version.
14
# This program is distributed in the hope that it will be useful,
15
# but WITHOUT ANY WARRANTY; without even the implied warranty of
16
# Mandos is distributed in the hope that it will be useful, but
17
# WITHOUT ANY WARRANTY; without even the implied warranty of
16
18
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17
19
# GNU General Public License for more details.
19
21
# You should have received a copy of the GNU General Public License
20
# along with this program. If not, see <http://www.gnu.org/licenses/>.
22
# Contact the authors at <mandos@fukt.bsnet.se>.
25
from __future__ import division, absolute_import, unicode_literals
22
# along with Mandos. If not, see <http://www.gnu.org/licenses/>.
24
# Contact the authors at <mandos@recompile.se>.
27
from __future__ import (division, absolute_import, print_function,
31
from future_builtins import *
29
from optparse import OptionParser
51
# Show warnings by default
52
if not sys.warnoptions:
54
warnings.simplefilter("default")
56
log = logging.getLogger(sys.argv[0])
57
logging.basicConfig(level="INFO", # Show info level messages
58
format="%(message)s") # Show basic log messages
60
logging.captureWarnings(True) # Show warnings via the logging system
62
if sys.version_info.major == 2:
35
65
locale.setlocale(locale.LC_ALL, "")
41
"LastCheckedOK": "Last Successful Check",
42
"LastApprovalRequest": "Last Approval Request",
44
"Interval": "Interval",
46
"Fingerprint": "Fingerprint",
47
"CheckerRunning": "Check Is Running",
48
"LastEnabled": "Last Enabled",
49
"ApprovalPending": "Approval Is Pending",
50
"ApprovedByDefault": "Approved By Default",
51
"ApprovalDelay": "Approval Delay",
52
"ApprovalDuration": "Approval Duration",
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
77
dbus.OBJECT_MANAGER_IFACE
78
except AttributeError:
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
83
parser = argparse.ArgumentParser()
85
add_command_line_options(parser)
87
options = parser.parse_args()
89
check_option_syntax(parser, options)
91
clientnames = options.client
94
log.setLevel(logging.DEBUG)
97
bus = dbus.SystemBus()
98
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
99
dbus_busname, server_dbus_path)
100
mandos_dbus_objc = bus.get_object(dbus_busname,
102
except dbus.exceptions.DBusException:
103
log.critical("Could not connect to Mandos server")
106
mandos_serv = dbus.Interface(mandos_dbus_objc,
107
dbus_interface=server_dbus_interface)
108
mandos_serv_object_manager = dbus.Interface(
109
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
112
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
113
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
114
with SilenceLogger("dbus.proxies"):
115
all_clients = {path: ifs_and_props[client_dbus_interface]
116
for path, ifs_and_props in
117
mandos_serv_object_manager
118
.GetManagedObjects().items()
119
if client_dbus_interface in ifs_and_props}
120
except dbus.exceptions.DBusException as e:
121
log.critical("Failed to access Mandos server through D-Bus:"
125
# Compile dict of (clients: properties) to process
129
clients = all_clients
131
for name in clientnames:
132
for objpath, properties in all_clients.items():
133
if properties["Name"] == name:
134
clients[objpath] = properties
137
log.critical("Client not found on server: %r", name)
140
# Run all commands on clients
141
commands = commands_from_options(options)
142
for command in commands:
143
command.run(clients, bus, mandos_serv)
146
def add_command_line_options(parser):
147
parser.add_argument("--version", action="version",
148
version="%(prog)s {}".format(version),
149
help="show version number and exit")
150
parser.add_argument("-a", "--all", action="store_true",
151
help="Select all clients")
152
parser.add_argument("-v", "--verbose", action="store_true",
153
help="Print all fields")
154
parser.add_argument("-j", "--dump-json", action="store_true",
155
help="Dump client data in JSON format")
156
enable_disable = parser.add_mutually_exclusive_group()
157
enable_disable.add_argument("-e", "--enable", action="store_true",
158
help="Enable client")
159
enable_disable.add_argument("-d", "--disable",
161
help="disable client")
162
parser.add_argument("-b", "--bump-timeout", action="store_true",
163
help="Bump timeout for client")
164
start_stop_checker = parser.add_mutually_exclusive_group()
165
start_stop_checker.add_argument("--start-checker",
167
help="Start checker for client")
168
start_stop_checker.add_argument("--stop-checker",
170
help="Stop checker for client")
171
parser.add_argument("-V", "--is-enabled", action="store_true",
172
help="Check if client is enabled")
173
parser.add_argument("-r", "--remove", action="store_true",
174
help="Remove client")
175
parser.add_argument("-c", "--checker",
176
help="Set checker command for client")
177
parser.add_argument("-t", "--timeout", type=string_to_delta,
178
help="Set timeout for client")
179
parser.add_argument("--extended-timeout", type=string_to_delta,
180
help="Set extended timeout for client")
181
parser.add_argument("-i", "--interval", type=string_to_delta,
182
help="Set checker interval for client")
183
approve_deny_default = parser.add_mutually_exclusive_group()
184
approve_deny_default.add_argument(
185
"--approve-by-default", action="store_true",
186
default=None, dest="approved_by_default",
187
help="Set client to be approved by default")
188
approve_deny_default.add_argument(
189
"--deny-by-default", action="store_false",
190
dest="approved_by_default",
191
help="Set client to be denied by default")
192
parser.add_argument("--approval-delay", type=string_to_delta,
193
help="Set delay before client approve/deny")
194
parser.add_argument("--approval-duration", type=string_to_delta,
195
help="Set duration of one client approval")
196
parser.add_argument("-H", "--host", help="Set host for client")
197
parser.add_argument("-s", "--secret",
198
type=argparse.FileType(mode="rb"),
199
help="Set password blob (file) for client")
200
approve_deny = parser.add_mutually_exclusive_group()
201
approve_deny.add_argument(
202
"-A", "--approve", action="store_true",
203
help="Approve any current client request")
204
approve_deny.add_argument("-D", "--deny", action="store_true",
205
help="Deny any current client request")
206
parser.add_argument("--debug", action="store_true",
207
help="Debug mode (show D-Bus commands)")
208
parser.add_argument("--check", action="store_true",
209
help="Run self-test")
210
parser.add_argument("client", nargs="*", help="Client name")
213
def string_to_delta(interval):
214
"""Parse a string and return a datetime.timedelta"""
217
return rfc3339_duration_to_delta(interval)
218
except ValueError as e:
219
log.warning("%s - Parsing as pre-1.6.1 interval instead",
221
return parse_pre_1_6_1_interval(interval)
224
def rfc3339_duration_to_delta(duration):
225
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
227
>>> rfc3339_duration_to_delta("P7D")
228
datetime.timedelta(7)
229
>>> rfc3339_duration_to_delta("PT60S")
230
datetime.timedelta(0, 60)
231
>>> rfc3339_duration_to_delta("PT60M")
232
datetime.timedelta(0, 3600)
233
>>> rfc3339_duration_to_delta("P60M")
234
datetime.timedelta(1680)
235
>>> rfc3339_duration_to_delta("PT24H")
236
datetime.timedelta(1)
237
>>> rfc3339_duration_to_delta("P1W")
238
datetime.timedelta(7)
239
>>> rfc3339_duration_to_delta("PT5M30S")
240
datetime.timedelta(0, 330)
241
>>> rfc3339_duration_to_delta("P1DT3M20S")
242
datetime.timedelta(1, 200)
243
>>> # Can not be empty:
244
>>> rfc3339_duration_to_delta("")
245
Traceback (most recent call last):
247
ValueError: Invalid RFC 3339 duration: u''
248
>>> # Must start with "P":
249
>>> rfc3339_duration_to_delta("1D")
250
Traceback (most recent call last):
252
ValueError: Invalid RFC 3339 duration: u'1D'
253
>>> # Must use correct order
254
>>> rfc3339_duration_to_delta("PT1S2M")
255
Traceback (most recent call last):
257
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
258
>>> # Time needs time marker
259
>>> rfc3339_duration_to_delta("P1H2S")
260
Traceback (most recent call last):
262
ValueError: Invalid RFC 3339 duration: u'P1H2S'
263
>>> # Weeks can not be combined with anything else
264
>>> rfc3339_duration_to_delta("P1D2W")
265
Traceback (most recent call last):
267
ValueError: Invalid RFC 3339 duration: u'P1D2W'
268
>>> rfc3339_duration_to_delta("P2W2H")
269
Traceback (most recent call last):
271
ValueError: Invalid RFC 3339 duration: u'P2W2H'
274
# Parsing an RFC 3339 duration with regular expressions is not
275
# possible - there would have to be multiple places for the same
276
# values, like seconds. The current code, while more esoteric, is
277
# cleaner without depending on a parsing library. If Python had a
278
# built-in library for parsing we would use it, but we'd like to
279
# avoid excessive use of external libraries.
281
# New type for defining tokens, syntax, and semantics all-in-one
282
Token = collections.namedtuple("Token", (
283
"regexp", # To match token; if "value" is not None, must have
284
# a "group" containing digits
285
"value", # datetime.timedelta or None
286
"followers")) # Tokens valid after this token
287
# RFC 3339 "duration" tokens, syntax, and semantics; taken from
288
# the "duration" ABNF definition in RFC 3339, Appendix A.
289
token_end = Token(re.compile(r"$"), None, frozenset())
290
token_second = Token(re.compile(r"(\d+)S"),
291
datetime.timedelta(seconds=1),
292
frozenset((token_end, )))
293
token_minute = Token(re.compile(r"(\d+)M"),
294
datetime.timedelta(minutes=1),
295
frozenset((token_second, token_end)))
296
token_hour = Token(re.compile(r"(\d+)H"),
297
datetime.timedelta(hours=1),
298
frozenset((token_minute, token_end)))
299
token_time = Token(re.compile(r"T"),
301
frozenset((token_hour, token_minute,
303
token_day = Token(re.compile(r"(\d+)D"),
304
datetime.timedelta(days=1),
305
frozenset((token_time, token_end)))
306
token_month = Token(re.compile(r"(\d+)M"),
307
datetime.timedelta(weeks=4),
308
frozenset((token_day, token_end)))
309
token_year = Token(re.compile(r"(\d+)Y"),
310
datetime.timedelta(weeks=52),
311
frozenset((token_month, token_end)))
312
token_week = Token(re.compile(r"(\d+)W"),
313
datetime.timedelta(weeks=1),
314
frozenset((token_end, )))
315
token_duration = Token(re.compile(r"P"), None,
316
frozenset((token_year, token_month,
317
token_day, token_time,
319
# Define starting values:
321
value = datetime.timedelta()
323
# Following valid tokens
324
followers = frozenset((token_duration, ))
325
# String left to parse
327
# Loop until end token is found
328
while found_token is not token_end:
329
# Search for any currently valid tokens
330
for token in followers:
331
match = token.regexp.match(s)
332
if match is not None:
334
if token.value is not None:
335
# Value found, parse digits
336
factor = int(match.group(1), 10)
337
# Add to value so far
338
value += factor * token.value
339
# Strip token from string
340
s = token.regexp.sub("", s, 1)
343
# Set valid next tokens
344
followers = found_token.followers
347
# No currently valid tokens were found
348
raise ValueError("Invalid RFC 3339 duration: {!r}"
354
def parse_pre_1_6_1_interval(interval):
355
"""Parse an interval string as documented by Mandos before 1.6.1,
356
and return a datetime.timedelta
358
>>> parse_pre_1_6_1_interval('7d')
359
datetime.timedelta(7)
360
>>> parse_pre_1_6_1_interval('60s')
361
datetime.timedelta(0, 60)
362
>>> parse_pre_1_6_1_interval('60m')
363
datetime.timedelta(0, 3600)
364
>>> parse_pre_1_6_1_interval('24h')
365
datetime.timedelta(1)
366
>>> parse_pre_1_6_1_interval('1w')
367
datetime.timedelta(7)
368
>>> parse_pre_1_6_1_interval('5m 30s')
369
datetime.timedelta(0, 330)
370
>>> parse_pre_1_6_1_interval('')
371
datetime.timedelta(0)
372
>>> # Ignore unknown characters, allow any order and repetitions
373
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
374
datetime.timedelta(2, 480, 18000)
378
value = datetime.timedelta(0)
379
regexp = re.compile(r"(\d+)([dsmhw]?)")
381
for num, suffix in regexp.findall(interval):
383
value += datetime.timedelta(int(num))
385
value += datetime.timedelta(0, int(num))
387
value += datetime.timedelta(0, 0, 0, 0, int(num))
389
value += datetime.timedelta(0, 0, 0, 0, 0, int(num))
391
value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))
393
value += datetime.timedelta(0, 0, 0, int(num))
397
def check_option_syntax(parser, options):
398
"""Apply additional restrictions on options, not expressible in
401
def has_actions(options):
402
return any((options.enable,
404
options.bump_timeout,
405
options.start_checker,
406
options.stop_checker,
409
options.checker is not None,
410
options.timeout is not None,
411
options.extended_timeout is not None,
412
options.interval is not None,
413
options.approved_by_default is not None,
414
options.approval_delay is not None,
415
options.approval_duration is not None,
416
options.host is not None,
417
options.secret is not None,
421
if has_actions(options) and not (options.client or options.all):
422
parser.error("Options require clients names or --all.")
423
if options.verbose and has_actions(options):
424
parser.error("--verbose can only be used alone.")
425
if options.dump_json and (options.verbose
426
or has_actions(options)):
427
parser.error("--dump-json can only be used alone.")
428
if options.all and not has_actions(options):
429
parser.error("--all requires an action.")
430
if options.is_enabled and len(options.client) > 1:
431
parser.error("--is-enabled requires exactly one client")
433
options.remove = False
434
if has_actions(options) and not options.deny:
435
parser.error("--remove can only be combined with --deny")
436
options.remove = True
439
class SilenceLogger(object):
440
"Simple context manager to silence a particular logger"
441
def __init__(self, loggername):
442
self.logger = logging.getLogger(loggername)
445
self.logger.addFilter(self.nullfilter)
448
class NullFilter(logging.Filter):
449
def filter(self, record):
452
nullfilter = NullFilter()
454
def __exit__(self, exc_type, exc_val, exc_tb):
455
self.logger.removeFilter(self.nullfilter)
458
def commands_from_options(options):
462
if options.is_enabled:
463
commands.append(IsEnabledCmd())
466
commands.append(ApproveCmd())
469
commands.append(DenyCmd())
472
commands.append(RemoveCmd())
474
if options.dump_json:
475
commands.append(DumpJSONCmd())
478
commands.append(EnableCmd())
481
commands.append(DisableCmd())
483
if options.bump_timeout:
484
commands.append(BumpTimeoutCmd())
486
if options.start_checker:
487
commands.append(StartCheckerCmd())
489
if options.stop_checker:
490
commands.append(StopCheckerCmd())
492
if options.approved_by_default is not None:
493
if options.approved_by_default:
494
commands.append(ApproveByDefaultCmd())
496
commands.append(DenyByDefaultCmd())
498
if options.checker is not None:
499
commands.append(SetCheckerCmd(options.checker))
501
if options.host is not None:
502
commands.append(SetHostCmd(options.host))
504
if options.secret is not None:
505
commands.append(SetSecretCmd(options.secret))
507
if options.timeout is not None:
508
commands.append(SetTimeoutCmd(options.timeout))
510
if options.extended_timeout:
512
SetExtendedTimeoutCmd(options.extended_timeout))
514
if options.interval is not None:
515
commands.append(SetIntervalCmd(options.interval))
517
if options.approval_delay is not None:
518
commands.append(SetApprovalDelayCmd(options.approval_delay))
520
if options.approval_duration is not None:
522
SetApprovalDurationCmd(options.approval_duration))
524
# If no command option has been given, show table of clients,
525
# optionally verbosely
527
commands.append(PrintTableCmd(verbose=options.verbose))
532
class Command(object):
533
"""Abstract class for commands"""
534
def run(self, clients, bus=None, mandos=None):
535
"""Normal commands should implement run_on_one_client(), but
536
commands which want to operate on all clients at the same time
537
can override this run() method instead."""
539
for clientpath, properties in clients.items():
540
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
541
dbus_busname, str(clientpath))
542
client = bus.get_object(dbus_busname, clientpath)
543
self.run_on_one_client(client, properties)
546
class IsEnabledCmd(Command):
547
def run(self, clients, bus=None, mandos=None):
548
client, properties = next(iter(clients.items()))
549
if self.is_enabled(client, properties):
552
def is_enabled(self, client, properties):
553
return properties["Enabled"]
556
class ApproveCmd(Command):
557
def run_on_one_client(self, client, properties):
558
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
559
client.__dbus_object_path__, client_dbus_interface)
560
client.Approve(dbus.Boolean(True),
561
dbus_interface=client_dbus_interface)
564
class DenyCmd(Command):
565
def run_on_one_client(self, client, properties):
566
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
567
client.__dbus_object_path__, client_dbus_interface)
568
client.Approve(dbus.Boolean(False),
569
dbus_interface=client_dbus_interface)
572
class RemoveCmd(Command):
573
def run_on_one_client(self, client, properties):
574
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
575
server_dbus_path, server_dbus_interface,
576
str(client.__dbus_object_path__))
577
self.mandos.RemoveClient(client.__dbus_object_path__)
580
class OutputCmd(Command):
581
"""Abstract class for commands outputting client details"""
582
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
583
"Created", "Interval", "Host", "KeyID",
584
"Fingerprint", "CheckerRunning", "LastEnabled",
585
"ApprovalPending", "ApprovedByDefault",
586
"LastApprovalRequest", "ApprovalDelay",
587
"ApprovalDuration", "Checker", "ExtendedTimeout",
588
"Expires", "LastCheckerStatus")
590
def run(self, clients, bus=None, mandos=None):
591
print(self.output(clients.values()))
593
def output(self, clients):
594
raise NotImplementedError()
597
class DumpJSONCmd(OutputCmd):
598
def output(self, clients):
599
data = {client["Name"]:
600
{key: self.dbus_boolean_to_bool(client[key])
601
for key in self.all_keywords}
602
for client in clients}
603
return json.dumps(data, indent=4, separators=(',', ': '))
606
def dbus_boolean_to_bool(value):
607
if isinstance(value, dbus.Boolean):
612
class PrintTableCmd(OutputCmd):
613
def __init__(self, verbose=False):
614
self.verbose = verbose
616
def output(self, clients):
617
default_keywords = ("Name", "Enabled", "Timeout",
619
keywords = default_keywords
621
keywords = self.all_keywords
622
return str(self.TableOfClients(clients, keywords))
624
class TableOfClients(object):
627
"Enabled": "Enabled",
628
"Timeout": "Timeout",
629
"LastCheckedOK": "Last Successful Check",
630
"LastApprovalRequest": "Last Approval Request",
631
"Created": "Created",
632
"Interval": "Interval",
634
"Fingerprint": "Fingerprint",
636
"CheckerRunning": "Check Is Running",
637
"LastEnabled": "Last Enabled",
638
"ApprovalPending": "Approval Is Pending",
639
"ApprovedByDefault": "Approved By Default",
640
"ApprovalDelay": "Approval Delay",
641
"ApprovalDuration": "Approval Duration",
642
"Checker": "Checker",
643
"ExtendedTimeout": "Extended Timeout",
644
"Expires": "Expires",
645
"LastCheckerStatus": "Last Checker Status",
648
def __init__(self, clients, keywords):
649
self.clients = clients
650
self.keywords = keywords
653
return "\n".join(self.rows())
655
if sys.version_info.major == 2:
656
__unicode__ = __str__
658
return str(self).encode(locale.getpreferredencoding())
661
format_string = self.row_formatting_string()
662
rows = [self.header_line(format_string)]
663
rows.extend(self.client_line(client, format_string)
664
for client in self.clients)
667
def row_formatting_string(self):
668
"Format string used to format table rows"
669
return " ".join("{{{key}:{width}}}".format(
670
width=max(len(self.tableheaders[key]),
671
*(len(self.string_from_client(client, key))
672
for client in self.clients)),
674
for key in self.keywords)
676
def string_from_client(self, client, key):
677
return self.valuetostring(client[key], key)
680
def valuetostring(cls, value, keyword):
681
if isinstance(value, dbus.Boolean):
682
return "Yes" if value else "No"
683
if keyword in ("Timeout", "Interval", "ApprovalDelay",
684
"ApprovalDuration", "ExtendedTimeout"):
685
return cls.milliseconds_to_string(value)
688
def header_line(self, format_string):
689
return format_string.format(**self.tableheaders)
691
def client_line(self, client, format_string):
692
return format_string.format(
693
**{key: self.string_from_client(client, key)
694
for key in self.keywords})
697
def milliseconds_to_string(ms):
698
td = datetime.timedelta(0, 0, 0, ms)
699
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
700
.format(days="{}T".format(td.days)
702
hours=td.seconds // 3600,
703
minutes=(td.seconds % 3600) // 60,
704
seconds=td.seconds % 60))
707
class PropertyCmd(Command):
708
"""Abstract class for Actions for setting one client property"""
710
def run_on_one_client(self, client, properties):
711
"""Set the Client's D-Bus property"""
712
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
713
client.__dbus_object_path__,
714
dbus.PROPERTIES_IFACE, client_dbus_interface,
715
self.propname, self.value_to_set
716
if not isinstance(self.value_to_set, dbus.Boolean)
717
else bool(self.value_to_set))
718
client.Set(client_dbus_interface, self.propname,
720
dbus_interface=dbus.PROPERTIES_IFACE)
724
raise NotImplementedError()
727
class EnableCmd(PropertyCmd):
729
value_to_set = dbus.Boolean(True)
732
class DisableCmd(PropertyCmd):
734
value_to_set = dbus.Boolean(False)
737
class BumpTimeoutCmd(PropertyCmd):
738
propname = "LastCheckedOK"
742
class StartCheckerCmd(PropertyCmd):
743
propname = "CheckerRunning"
744
value_to_set = dbus.Boolean(True)
747
class StopCheckerCmd(PropertyCmd):
748
propname = "CheckerRunning"
749
value_to_set = dbus.Boolean(False)
752
class ApproveByDefaultCmd(PropertyCmd):
753
propname = "ApprovedByDefault"
754
value_to_set = dbus.Boolean(True)
757
class DenyByDefaultCmd(PropertyCmd):
758
propname = "ApprovedByDefault"
759
value_to_set = dbus.Boolean(False)
762
class PropertyValueCmd(PropertyCmd):
763
"""Abstract class for PropertyCmd recieving a value as argument"""
764
def __init__(self, value):
765
self.value_to_set = value
768
class SetCheckerCmd(PropertyValueCmd):
772
class SetHostCmd(PropertyValueCmd):
776
class SetSecretCmd(PropertyValueCmd):
780
def value_to_set(self):
784
def value_to_set(self, value):
785
"""When setting, read data from supplied file object"""
786
self._vts = value.read()
790
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
791
"""Abstract class for PropertyValueCmd taking a value argument as
792
a datetime.timedelta() but should store it as milliseconds."""
795
def value_to_set(self):
799
def value_to_set(self, value):
800
"""When setting, convert value from a datetime.timedelta"""
801
self._vts = int(round(value.total_seconds() * 1000))
804
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
808
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
809
propname = "ExtendedTimeout"
812
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
813
propname = "Interval"
816
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
817
propname = "ApprovalDelay"
820
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
821
propname = "ApprovalDuration"
825
class Test_string_to_delta(unittest.TestCase):
826
def test_handles_basic_rfc3339(self):
827
self.assertEqual(string_to_delta("PT0S"),
828
datetime.timedelta())
829
self.assertEqual(string_to_delta("P0D"),
830
datetime.timedelta())
831
self.assertEqual(string_to_delta("PT1S"),
832
datetime.timedelta(0, 1))
833
self.assertEqual(string_to_delta("PT2H"),
834
datetime.timedelta(0, 7200))
836
def test_falls_back_to_pre_1_6_1_with_warning(self):
837
# assertLogs only exists in Python 3.4
838
if hasattr(self, "assertLogs"):
839
with self.assertLogs(log, logging.WARNING):
840
value = string_to_delta("2h")
842
class WarningFilter(logging.Filter):
843
"""Don't show, but record the presence of, warnings"""
844
def filter(self, record):
845
is_warning = record.levelno >= logging.WARNING
846
self.found = is_warning or getattr(self, "found",
848
return not is_warning
849
warning_filter = WarningFilter()
850
log.addFilter(warning_filter)
852
value = string_to_delta("2h")
854
log.removeFilter(warning_filter)
855
self.assertTrue(getattr(warning_filter, "found", False))
856
self.assertEqual(value, datetime.timedelta(0, 7200))
859
class Test_check_option_syntax(unittest.TestCase):
861
self.parser = argparse.ArgumentParser()
862
add_command_line_options(self.parser)
864
def test_actions_requires_client_or_all(self):
865
for action, value in self.actions.items():
866
options = self.parser.parse_args()
867
setattr(options, action, value)
868
with self.assertParseError():
869
self.check_option_syntax(options)
871
# This mostly corresponds to the definition from has_actions() in
872
# check_option_syntax()
874
# The actual values set here are not that important, but we do
875
# at least stick to the correct types, even though they are
879
"bump_timeout": True,
880
"start_checker": True,
881
"stop_checker": True,
885
"timeout": datetime.timedelta(),
886
"extended_timeout": datetime.timedelta(),
887
"interval": datetime.timedelta(),
888
"approved_by_default": True,
889
"approval_delay": datetime.timedelta(),
890
"approval_duration": datetime.timedelta(),
892
"secret": io.BytesIO(b"x"),
55
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
56
domain = "se.bsnet.fukt"
57
busname = domain + ".Mandos"
59
server_interface = domain + ".Mandos"
60
client_interface = domain + ".Mandos.Client"
63
def timedelta_to_milliseconds(td):
64
"""Convert a datetime.timedelta object to milliseconds"""
65
return ((td.days * 24 * 60 * 60 * 1000)
67
+ (td.microseconds // 1000))
69
def milliseconds_to_string(ms):
70
td = datetime.timedelta(0, 0, 0, ms)
71
return ("%(days)s%(hours)02d:%(minutes)02d:%(seconds)02d"
72
% { "days": "%dT" % td.days if td.days else "",
73
"hours": td.seconds // 3600,
74
"minutes": (td.seconds % 3600) // 60,
75
"seconds": td.seconds % 60,
79
def string_to_delta(interval):
80
"""Parse a string and return a datetime.timedelta
82
>>> string_to_delta("7d")
84
>>> string_to_delta("60s")
85
datetime.timedelta(0, 60)
86
>>> string_to_delta("60m")
87
datetime.timedelta(0, 3600)
88
>>> string_to_delta("24h")
90
>>> string_to_delta("1w")
92
>>> string_to_delta("5m 30s")
93
datetime.timedelta(0, 330)
95
timevalue = datetime.timedelta(0)
96
regexp = re.compile("\d+[dsmhw]")
98
for s in regexp.findall(interval):
100
suffix = unicode(s[-1])
103
delta = datetime.timedelta(value)
105
delta = datetime.timedelta(0, value)
107
delta = datetime.timedelta(0, 0, 0, 0, value)
109
delta = datetime.timedelta(0, 0, 0, 0, 0, value)
111
delta = datetime.timedelta(0, 0, 0, 0, 0, 0, value)
114
except (ValueError, IndexError):
119
def print_clients(clients, keywords):
120
def valuetostring(value, keyword):
121
if type(value) is dbus.Boolean:
122
return "Yes" if value else "No"
123
if keyword in ("Timeout", "Interval", "ApprovalDelay",
125
return milliseconds_to_string(value)
126
return unicode(value)
128
# Create format string to print table rows
129
format_string = " ".join("%%-%ds" %
130
max(len(tablewords[key]),
131
max(len(valuetostring(client[key],
137
print format_string % tuple(tablewords[key] for key in keywords)
138
for client in clients:
139
print format_string % tuple(valuetostring(client[key], key)
142
def has_actions(options):
143
return any((options.enable,
145
options.bump_timeout,
146
options.start_checker,
147
options.stop_checker,
150
options.checker is not None,
151
options.timeout is not None,
152
options.interval is not None,
153
options.approved_by_default is not None,
154
options.approval_delay is not None,
155
options.approval_duration is not None,
156
options.host is not None,
157
options.secret is not None,
162
parser = OptionParser(version = "%%prog %s" % version)
163
parser.add_option("-a", "--all", action="store_true",
164
help="Select all clients")
165
parser.add_option("-v", "--verbose", action="store_true",
166
help="Print all fields")
167
parser.add_option("-e", "--enable", action="store_true",
168
help="Enable client")
169
parser.add_option("-d", "--disable", action="store_true",
170
help="disable client")
171
parser.add_option("-b", "--bump-timeout", action="store_true",
172
help="Bump timeout for client")
173
parser.add_option("--start-checker", action="store_true",
174
help="Start checker for client")
175
parser.add_option("--stop-checker", action="store_true",
176
help="Stop checker for client")
177
parser.add_option("-V", "--is-enabled", action="store_true",
178
help="Check if client is enabled")
179
parser.add_option("-r", "--remove", action="store_true",
180
help="Remove client")
181
parser.add_option("-c", "--checker", type="string",
182
help="Set checker command for client")
183
parser.add_option("-t", "--timeout", type="string",
184
help="Set timeout for client")
185
parser.add_option("-i", "--interval", type="string",
186
help="Set checker interval for client")
187
parser.add_option("--approve-by-default", action="store_true",
188
dest="approved_by_default",
189
help="Set client to be approved by default")
190
parser.add_option("--deny-by-default", action="store_false",
191
dest="approved_by_default",
192
help="Set client to be denied by default")
193
parser.add_option("--approval-delay", type="string",
194
help="Set delay before client approve/deny")
195
parser.add_option("--approval-duration", type="string",
196
help="Set duration of one client approval")
197
parser.add_option("-H", "--host", type="string",
198
help="Set host for client")
199
parser.add_option("-s", "--secret", type="string",
200
help="Set password blob (file) for client")
201
parser.add_option("-A", "--approve", action="store_true",
202
help="Approve any current client request")
203
parser.add_option("-D", "--deny", action="store_true",
204
help="Deny any current client request")
205
options, client_names = parser.parse_args()
207
if has_actions(options) and not client_names and not options.all:
208
parser.error("Options require clients names or --all.")
209
if options.verbose and has_actions(options):
210
parser.error("--verbose can only be used alone or with"
212
if options.all and not has_actions(options):
213
parser.error("--all requires an action.")
216
bus = dbus.SystemBus()
217
mandos_dbus_objc = bus.get_object(busname, server_path)
218
except dbus.exceptions.DBusException:
219
print >> sys.stderr, "Could not connect to Mandos server"
222
mandos_serv = dbus.Interface(mandos_dbus_objc,
223
dbus_interface = server_interface)
225
#block stderr since dbus library prints to stderr
897
@contextlib.contextmanager
898
def assertParseError(self):
899
with self.assertRaises(SystemExit) as e:
900
with self.temporarily_suppress_stderr():
902
# Exit code from argparse is guaranteed to be "2". Reference:
903
# https://docs.python.org/3/library
904
# /argparse.html#exiting-methods
905
self.assertEqual(e.exception.code, 2)
908
@contextlib.contextmanager
909
def temporarily_suppress_stderr():
226
910
null = os.open(os.path.devnull, os.O_RDWR)
227
911
stderrcopy = os.dup(sys.stderr.fileno())
228
912
os.dup2(null, sys.stderr.fileno())
232
mandos_clients = mandos_serv.GetAllClientsWithProperties()
235
os.dup2(stderrcopy, sys.stderr.fileno())
237
except dbus.exceptions.DBusException, e:
238
print >> sys.stderr, "Access denied: Accessing mandos server through dbus."
241
# Compile dict of (clients: properties) to process
244
if options.all or not client_names:
245
clients = dict((bus.get_object(busname, path), properties)
246
for path, properties in
247
mandos_clients.iteritems())
249
for name in client_names:
250
for path, client in mandos_clients.iteritems():
251
if client["Name"] == name:
252
client_objc = bus.get_object(busname, path)
253
clients[client_objc] = client
256
print >> sys.stderr, "Client not found on server: %r" % name
259
if not has_actions(options) and clients:
261
keywords = ("Name", "Enabled", "Timeout",
262
"LastCheckedOK", "Created", "Interval",
263
"Host", "Fingerprint", "CheckerRunning",
264
"LastEnabled", "ApprovalPending",
266
"LastApprovalRequest", "ApprovalDelay",
267
"ApprovalDuration", "Checker")
269
keywords = defaultkeywords
271
print_clients(clients.values(), keywords)
273
# Process each client in the list by all selected options
274
for client in clients:
276
mandos_serv.RemoveClient(client.__dbus_object_path__)
278
client.Enable(dbus_interface=client_interface)
280
client.Disable(dbus_interface=client_interface)
281
if options.bump_timeout:
282
client.CheckedOK(dbus_interface=client_interface)
283
if options.start_checker:
284
client.StartChecker(dbus_interface=client_interface)
285
if options.stop_checker:
286
client.StopChecker(dbus_interface=client_interface)
287
if options.is_enabled:
288
sys.exit(0 if client.Get(client_interface,
290
dbus_interface=dbus.PROPERTIES_IFACE)
293
client.Set(client_interface, "Checker", options.checker,
294
dbus_interface=dbus.PROPERTIES_IFACE)
296
client.Set(client_interface, "Host", options.host,
297
dbus_interface=dbus.PROPERTIES_IFACE)
299
client.Set(client_interface, "Interval",
300
timedelta_to_milliseconds
301
(string_to_delta(options.interval)),
302
dbus_interface=dbus.PROPERTIES_IFACE)
303
if options.approval_delay:
304
client.Set(client_interface, "ApprovalDelay",
305
timedelta_to_milliseconds
306
(string_to_delta(options.
308
dbus_interface=dbus.PROPERTIES_IFACE)
309
if options.approval_duration:
310
client.Set(client_interface, "ApprovalDuration",
311
timedelta_to_milliseconds
312
(string_to_delta(options.
314
dbus_interface=dbus.PROPERTIES_IFACE)
316
client.Set(client_interface, "Timeout",
317
timedelta_to_milliseconds
318
(string_to_delta(options.timeout)),
319
dbus_interface=dbus.PROPERTIES_IFACE)
321
client.Set(client_interface, "Secret",
322
dbus.ByteArray(open(options.secret,
324
dbus_interface=dbus.PROPERTIES_IFACE)
325
if options.approved_by_default is not None:
326
client.Set(client_interface, "ApprovedByDefault",
328
.approved_by_default),
329
dbus_interface=dbus.PROPERTIES_IFACE)
331
client.Approve(dbus.Boolean(True),
332
dbus_interface=client_interface)
334
client.Approve(dbus.Boolean(False),
335
dbus_interface=client_interface)
918
os.dup2(stderrcopy, sys.stderr.fileno())
921
def check_option_syntax(self, options):
922
check_option_syntax(self.parser, options)
924
def test_actions_conflicts_with_verbose(self):
925
for action, value in self.actions.items():
926
options = self.parser.parse_args()
927
setattr(options, action, value)
928
options.verbose = True
929
with self.assertParseError():
930
self.check_option_syntax(options)
932
def test_dump_json_conflicts_with_verbose(self):
933
options = self.parser.parse_args()
934
options.dump_json = True
935
options.verbose = True
936
with self.assertParseError():
937
self.check_option_syntax(options)
939
def test_dump_json_conflicts_with_action(self):
940
for action, value in self.actions.items():
941
options = self.parser.parse_args()
942
setattr(options, action, value)
943
options.dump_json = True
944
with self.assertParseError():
945
self.check_option_syntax(options)
947
def test_all_can_not_be_alone(self):
948
options = self.parser.parse_args()
950
with self.assertParseError():
951
self.check_option_syntax(options)
953
def test_all_is_ok_with_any_action(self):
954
for action, value in self.actions.items():
955
options = self.parser.parse_args()
956
setattr(options, action, value)
958
self.check_option_syntax(options)
960
def test_is_enabled_fails_without_client(self):
961
options = self.parser.parse_args()
962
options.is_enabled = True
963
with self.assertParseError():
964
self.check_option_syntax(options)
966
def test_is_enabled_works_with_one_client(self):
967
options = self.parser.parse_args()
968
options.is_enabled = True
969
options.client = ["foo"]
970
self.check_option_syntax(options)
972
def test_is_enabled_fails_with_two_clients(self):
973
options = self.parser.parse_args()
974
options.is_enabled = True
975
options.client = ["foo", "barbar"]
976
with self.assertParseError():
977
self.check_option_syntax(options)
979
def test_remove_can_only_be_combined_with_action_deny(self):
980
for action, value in self.actions.items():
981
if action in {"remove", "deny"}:
983
options = self.parser.parse_args()
984
setattr(options, action, value)
986
options.remove = True
987
with self.assertParseError():
988
self.check_option_syntax(options)
991
class Test_SilenceLogger(unittest.TestCase):
992
loggername = "mandos-ctl.Test_SilenceLogger"
993
log = logging.getLogger(loggername)
994
log.propagate = False
995
log.addHandler(logging.NullHandler())
998
self.counting_filter = self.CountingFilter()
1000
class CountingFilter(logging.Filter):
1001
"Count number of records"
1003
def filter(self, record):
1007
def test_should_filter_records_only_when_active(self):
1009
with SilenceLogger(self.loggername):
1010
self.log.addFilter(self.counting_filter)
1011
self.log.info("Filtered log message 1")
1012
self.log.info("Non-filtered message 2")
1013
self.log.info("Non-filtered message 3")
1015
self.log.removeFilter(self.counting_filter)
1016
self.assertEqual(self.counting_filter.count, 2)
1019
class Test_commands_from_options(unittest.TestCase):
1021
self.parser = argparse.ArgumentParser()
1022
add_command_line_options(self.parser)
1024
def test_is_enabled(self):
1025
self.assert_command_from_args(["--is-enabled", "foo"],
1028
def assert_command_from_args(self, args, command_cls,
1030
"""Assert that parsing ARGS should result in an instance of
1031
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1032
options = self.parser.parse_args(args)
1033
check_option_syntax(self.parser, options)
1034
commands = commands_from_options(options)
1035
self.assertEqual(len(commands), 1)
1036
command = commands[0]
1037
self.assertIsInstance(command, command_cls)
1038
for key, value in cmd_attrs.items():
1039
self.assertEqual(getattr(command, key), value)
1041
def test_is_enabled_short(self):
1042
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1044
def test_approve(self):
1045
self.assert_command_from_args(["--approve", "foo"],
1048
def test_approve_short(self):
1049
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1051
def test_deny(self):
1052
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1054
def test_deny_short(self):
1055
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1057
def test_remove(self):
1058
self.assert_command_from_args(["--remove", "foo"],
1061
def test_deny_before_remove(self):
1062
options = self.parser.parse_args(["--deny", "--remove",
1064
check_option_syntax(self.parser, options)
1065
commands = commands_from_options(options)
1066
self.assertEqual(len(commands), 2)
1067
self.assertIsInstance(commands[0], DenyCmd)
1068
self.assertIsInstance(commands[1], RemoveCmd)
1070
def test_deny_before_remove_reversed(self):
1071
options = self.parser.parse_args(["--remove", "--deny",
1073
check_option_syntax(self.parser, options)
1074
commands = commands_from_options(options)
1075
self.assertEqual(len(commands), 2)
1076
self.assertIsInstance(commands[0], DenyCmd)
1077
self.assertIsInstance(commands[1], RemoveCmd)
1079
def test_remove_short(self):
1080
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1082
def test_dump_json(self):
1083
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1085
def test_enable(self):
1086
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1088
def test_enable_short(self):
1089
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1091
def test_disable(self):
1092
self.assert_command_from_args(["--disable", "foo"],
1095
def test_disable_short(self):
1096
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1098
def test_bump_timeout(self):
1099
self.assert_command_from_args(["--bump-timeout", "foo"],
1102
def test_bump_timeout_short(self):
1103
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1105
def test_start_checker(self):
1106
self.assert_command_from_args(["--start-checker", "foo"],
1109
def test_stop_checker(self):
1110
self.assert_command_from_args(["--stop-checker", "foo"],
1113
def test_approve_by_default(self):
1114
self.assert_command_from_args(["--approve-by-default", "foo"],
1115
ApproveByDefaultCmd)
1117
def test_deny_by_default(self):
1118
self.assert_command_from_args(["--deny-by-default", "foo"],
1121
def test_checker(self):
1122
self.assert_command_from_args(["--checker", ":", "foo"],
1123
SetCheckerCmd, value_to_set=":")
1125
def test_checker_empty(self):
1126
self.assert_command_from_args(["--checker", "", "foo"],
1127
SetCheckerCmd, value_to_set="")
1129
def test_checker_short(self):
1130
self.assert_command_from_args(["-c", ":", "foo"],
1131
SetCheckerCmd, value_to_set=":")
1133
def test_host(self):
1134
self.assert_command_from_args(["--host", "foo.example.org",
1136
value_to_set="foo.example.org")
1138
def test_host_short(self):
1139
self.assert_command_from_args(["-H", "foo.example.org",
1141
value_to_set="foo.example.org")
1143
def test_secret_devnull(self):
1144
self.assert_command_from_args(["--secret", os.path.devnull,
1145
"foo"], SetSecretCmd,
1148
def test_secret_tempfile(self):
1149
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1150
value = b"secret\0xyzzy\nbar"
1153
self.assert_command_from_args(["--secret", f.name,
1154
"foo"], SetSecretCmd,
1157
def test_secret_devnull_short(self):
1158
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1159
SetSecretCmd, value_to_set=b"")
1161
def test_secret_tempfile_short(self):
1162
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1163
value = b"secret\0xyzzy\nbar"
1166
self.assert_command_from_args(["-s", f.name, "foo"],
1170
def test_timeout(self):
1171
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1173
value_to_set=300000)
1175
def test_timeout_short(self):
1176
self.assert_command_from_args(["-t", "PT5M", "foo"],
1178
value_to_set=300000)
1180
def test_extended_timeout(self):
1181
self.assert_command_from_args(["--extended-timeout", "PT15M",
1183
SetExtendedTimeoutCmd,
1184
value_to_set=900000)
1186
def test_interval(self):
1187
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1189
value_to_set=120000)
1191
def test_interval_short(self):
1192
self.assert_command_from_args(["-i", "PT2M", "foo"],
1194
value_to_set=120000)
1196
def test_approval_delay(self):
1197
self.assert_command_from_args(["--approval-delay", "PT30S",
1198
"foo"], SetApprovalDelayCmd,
1201
def test_approval_duration(self):
1202
self.assert_command_from_args(["--approval-duration", "PT1S",
1203
"foo"], SetApprovalDurationCmd,
1206
def test_print_table(self):
1207
self.assert_command_from_args([], PrintTableCmd,
1210
def test_print_table_verbose(self):
1211
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1214
def test_print_table_verbose_short(self):
1215
self.assert_command_from_args(["-v"], PrintTableCmd,
1219
class TestCmd(unittest.TestCase):
1220
"""Abstract class for tests of command classes"""
1224
class MockClient(object):
1225
def __init__(self, name, **attributes):
1226
self.__dbus_object_path__ = "/clients/{}".format(name)
1227
self.attributes = attributes
1228
self.attributes["Name"] = name
1230
def Set(self, interface, propname, value, dbus_interface):
1231
testcase.assertEqual(interface, client_dbus_interface)
1232
testcase.assertEqual(dbus_interface,
1233
dbus.PROPERTIES_IFACE)
1234
self.attributes[propname] = value
1235
def Get(self, interface, propname, dbus_interface):
1236
testcase.assertEqual(interface, client_dbus_interface)
1237
testcase.assertEqual(dbus_interface,
1238
dbus.PROPERTIES_IFACE)
1239
return self.attributes[propname]
1240
def Approve(self, approve, dbus_interface):
1241
testcase.assertEqual(dbus_interface,
1242
client_dbus_interface)
1243
self.calls.append(("Approve", (approve,
1245
self.client = MockClient(
1247
KeyID=("92ed150794387c03ce684574b1139a65"
1248
"94a34f895daaaf09fd8ea90a27cddb12"),
1250
Host="foo.example.org",
1251
Enabled=dbus.Boolean(True),
1253
LastCheckedOK="2019-02-03T00:00:00",
1254
Created="2019-01-02T00:00:00",
1256
Fingerprint=("778827225BA7DE539C5A"
1257
"7CFA59CFF7CDBD9A5920"),
1258
CheckerRunning=dbus.Boolean(False),
1259
LastEnabled="2019-01-03T00:00:00",
1260
ApprovalPending=dbus.Boolean(False),
1261
ApprovedByDefault=dbus.Boolean(True),
1262
LastApprovalRequest="",
1264
ApprovalDuration=1000,
1265
Checker="fping -q -- %(host)s",
1266
ExtendedTimeout=900000,
1267
Expires="2019-02-04T00:00:00",
1268
LastCheckerStatus=0)
1269
self.other_client = MockClient(
1271
KeyID=("0558568eedd67d622f5c83b35a115f79"
1272
"6ab612cff5ad227247e46c2b020f441c"),
1273
Secret=b"secretbar",
1275
Enabled=dbus.Boolean(True),
1277
LastCheckedOK="2019-02-04T00:00:00",
1278
Created="2019-01-03T00:00:00",
1280
Fingerprint=("3E393AEAEFB84C7E89E2"
1281
"F547B3A107558FCA3A27"),
1282
CheckerRunning=dbus.Boolean(True),
1283
LastEnabled="2019-01-04T00:00:00",
1284
ApprovalPending=dbus.Boolean(False),
1285
ApprovedByDefault=dbus.Boolean(False),
1286
LastApprovalRequest="2019-01-03T00:00:00",
1287
ApprovalDelay=30000,
1288
ApprovalDuration=93785000,
1290
ExtendedTimeout=900000,
1291
Expires="2019-02-05T00:00:00",
1292
LastCheckerStatus=-2)
1293
self.clients = collections.OrderedDict(
1295
("/clients/foo", self.client.attributes),
1296
("/clients/barbar", self.other_client.attributes),
1298
self.one_client = {"/clients/foo": self.client.attributes}
1304
def get_object(client_bus_name, path):
1305
self.assertEqual(client_bus_name, dbus_busname)
1307
# Note: "self" here is the TestCmd instance, not
1308
# the Bus instance, since this is a static method!
1309
"/clients/foo": self.client,
1310
"/clients/barbar": self.other_client,
1315
class TestIsEnabledCmd(TestCmd):
1316
def test_is_enabled(self):
1317
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1319
for client, properties
1320
in self.clients.items()))
1322
def test_is_enabled_run_exits_successfully(self):
1323
with self.assertRaises(SystemExit) as e:
1324
IsEnabledCmd().run(self.one_client)
1325
if e.exception.code is not None:
1326
self.assertEqual(e.exception.code, 0)
1328
self.assertIsNone(e.exception.code)
1330
def test_is_enabled_run_exits_with_failure(self):
1331
self.client.attributes["Enabled"] = dbus.Boolean(False)
1332
with self.assertRaises(SystemExit) as e:
1333
IsEnabledCmd().run(self.one_client)
1334
if isinstance(e.exception.code, int):
1335
self.assertNotEqual(e.exception.code, 0)
1337
self.assertIsNotNone(e.exception.code)
1340
class TestApproveCmd(TestCmd):
1341
def test_approve(self):
1342
ApproveCmd().run(self.clients, self.bus)
1343
for clientpath in self.clients:
1344
client = self.bus.get_object(dbus_busname, clientpath)
1345
self.assertIn(("Approve", (True, client_dbus_interface)),
1349
class TestDenyCmd(TestCmd):
1350
def test_deny(self):
1351
DenyCmd().run(self.clients, self.bus)
1352
for clientpath in self.clients:
1353
client = self.bus.get_object(dbus_busname, clientpath)
1354
self.assertIn(("Approve", (False, client_dbus_interface)),
1358
class TestRemoveCmd(TestCmd):
1359
def test_remove(self):
1360
class MockMandos(object):
1363
def RemoveClient(self, dbus_path):
1364
self.calls.append(("RemoveClient", (dbus_path,)))
1365
mandos = MockMandos()
1366
super(TestRemoveCmd, self).setUp()
1367
RemoveCmd().run(self.clients, self.bus, mandos)
1368
self.assertEqual(len(mandos.calls), 2)
1369
for clientpath in self.clients:
1370
self.assertIn(("RemoveClient", (clientpath,)),
1374
class TestDumpJSONCmd(TestCmd):
1376
self.expected_json = {
1379
"KeyID": ("92ed150794387c03ce684574b1139a65"
1380
"94a34f895daaaf09fd8ea90a27cddb12"),
1381
"Host": "foo.example.org",
1384
"LastCheckedOK": "2019-02-03T00:00:00",
1385
"Created": "2019-01-02T00:00:00",
1387
"Fingerprint": ("778827225BA7DE539C5A"
1388
"7CFA59CFF7CDBD9A5920"),
1389
"CheckerRunning": False,
1390
"LastEnabled": "2019-01-03T00:00:00",
1391
"ApprovalPending": False,
1392
"ApprovedByDefault": True,
1393
"LastApprovalRequest": "",
1395
"ApprovalDuration": 1000,
1396
"Checker": "fping -q -- %(host)s",
1397
"ExtendedTimeout": 900000,
1398
"Expires": "2019-02-04T00:00:00",
1399
"LastCheckerStatus": 0,
1403
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1404
"6ab612cff5ad227247e46c2b020f441c"),
1405
"Host": "192.0.2.3",
1408
"LastCheckedOK": "2019-02-04T00:00:00",
1409
"Created": "2019-01-03T00:00:00",
1411
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1412
"F547B3A107558FCA3A27"),
1413
"CheckerRunning": True,
1414
"LastEnabled": "2019-01-04T00:00:00",
1415
"ApprovalPending": False,
1416
"ApprovedByDefault": False,
1417
"LastApprovalRequest": "2019-01-03T00:00:00",
1418
"ApprovalDelay": 30000,
1419
"ApprovalDuration": 93785000,
1421
"ExtendedTimeout": 900000,
1422
"Expires": "2019-02-05T00:00:00",
1423
"LastCheckerStatus": -2,
1426
return super(TestDumpJSONCmd, self).setUp()
1428
def test_normal(self):
1429
output = DumpJSONCmd().output(self.clients.values())
1430
json_data = json.loads(output)
1431
self.assertDictEqual(json_data, self.expected_json)
1433
def test_one_client(self):
1434
output = DumpJSONCmd().output(self.one_client.values())
1435
json_data = json.loads(output)
1436
expected_json = {"foo": self.expected_json["foo"]}
1437
self.assertDictEqual(json_data, expected_json)
1440
class TestPrintTableCmd(TestCmd):
1441
def test_normal(self):
1442
output = PrintTableCmd().output(self.clients.values())
1443
expected_output = "\n".join((
1444
"Name Enabled Timeout Last Successful Check",
1445
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1446
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1448
self.assertEqual(output, expected_output)
1450
def test_verbose(self):
1451
output = PrintTableCmd(verbose=True).output(
1452
self.clients.values())
1467
"Last Successful Check ",
1468
"2019-02-03T00:00:00 ",
1469
"2019-02-04T00:00:00 ",
1472
"2019-01-02T00:00:00 ",
1473
"2019-01-03T00:00:00 ",
1485
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1487
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1491
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1492
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1494
"Check Is Running ",
1499
"2019-01-03T00:00:00 ",
1500
"2019-01-04T00:00:00 ",
1502
"Approval Is Pending ",
1506
"Approved By Default ",
1510
"Last Approval Request ",
1512
"2019-01-03T00:00:00 ",
1518
"Approval Duration ",
1523
"fping -q -- %(host)s ",
1526
"Extended Timeout ",
1531
"2019-02-04T00:00:00 ",
1532
"2019-02-05T00:00:00 ",
1534
"Last Checker Status",
1539
num_lines = max(len(rows) for rows in columns)
1540
expected_output = "\n".join("".join(rows[line]
1541
for rows in columns)
1542
for line in range(num_lines))
1543
self.assertEqual(output, expected_output)
1545
def test_one_client(self):
1546
output = PrintTableCmd().output(self.one_client.values())
1547
expected_output = "\n".join((
1548
"Name Enabled Timeout Last Successful Check",
1549
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1551
self.assertEqual(output, expected_output)
1554
class TestPropertyCmd(TestCmd):
1555
"""Abstract class for tests of PropertyCmd classes"""
1557
if not hasattr(self, "command"):
1559
values_to_get = getattr(self, "values_to_get",
1561
for value_to_set, value_to_get in zip(self.values_to_set,
1563
for clientpath in self.clients:
1564
client = self.bus.get_object(dbus_busname, clientpath)
1565
old_value = client.attributes[self.propname]
1566
self.assertNotIsInstance(old_value, self.Unique)
1567
client.attributes[self.propname] = self.Unique()
1568
self.run_command(value_to_set, self.clients)
1569
for clientpath in self.clients:
1570
client = self.bus.get_object(dbus_busname, clientpath)
1571
value = client.attributes[self.propname]
1572
self.assertNotIsInstance(value, self.Unique)
1573
self.assertEqual(value, value_to_get)
1575
class Unique(object):
1576
"""Class for objects which exist only to be unique objects,
1577
since unittest.mock.sentinel only exists in Python 3.3"""
1579
def run_command(self, value, clients):
1580
self.command().run(clients, self.bus)
1583
class TestEnableCmd(TestPropertyCmd):
1585
propname = "Enabled"
1586
values_to_set = [dbus.Boolean(True)]
1589
class TestDisableCmd(TestPropertyCmd):
1590
command = DisableCmd
1591
propname = "Enabled"
1592
values_to_set = [dbus.Boolean(False)]
1595
class TestBumpTimeoutCmd(TestPropertyCmd):
1596
command = BumpTimeoutCmd
1597
propname = "LastCheckedOK"
1598
values_to_set = [""]
1601
class TestStartCheckerCmd(TestPropertyCmd):
1602
command = StartCheckerCmd
1603
propname = "CheckerRunning"
1604
values_to_set = [dbus.Boolean(True)]
1607
class TestStopCheckerCmd(TestPropertyCmd):
1608
command = StopCheckerCmd
1609
propname = "CheckerRunning"
1610
values_to_set = [dbus.Boolean(False)]
1613
class TestApproveByDefaultCmd(TestPropertyCmd):
1614
command = ApproveByDefaultCmd
1615
propname = "ApprovedByDefault"
1616
values_to_set = [dbus.Boolean(True)]
1619
class TestDenyByDefaultCmd(TestPropertyCmd):
1620
command = DenyByDefaultCmd
1621
propname = "ApprovedByDefault"
1622
values_to_set = [dbus.Boolean(False)]
1625
class TestPropertyValueCmd(TestPropertyCmd):
1626
"""Abstract class for tests of PropertyValueCmd classes"""
1629
if type(self) is TestPropertyValueCmd:
1631
return super(TestPropertyValueCmd, self).runTest()
1633
def run_command(self, value, clients):
1634
self.command(value).run(clients, self.bus)
1637
class TestSetCheckerCmd(TestPropertyValueCmd):
1638
command = SetCheckerCmd
1639
propname = "Checker"
1640
values_to_set = ["", ":", "fping -q -- %s"]
1643
class TestSetHostCmd(TestPropertyValueCmd):
1644
command = SetHostCmd
1646
values_to_set = ["192.0.2.3", "foo.example.org"]
1649
class TestSetSecretCmd(TestPropertyValueCmd):
1650
command = SetSecretCmd
1652
values_to_set = [io.BytesIO(b""),
1653
io.BytesIO(b"secret\0xyzzy\nbar")]
1654
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1657
class TestSetTimeoutCmd(TestPropertyValueCmd):
1658
command = SetTimeoutCmd
1659
propname = "Timeout"
1660
values_to_set = [datetime.timedelta(),
1661
datetime.timedelta(minutes=5),
1662
datetime.timedelta(seconds=1),
1663
datetime.timedelta(weeks=1),
1664
datetime.timedelta(weeks=52)]
1665
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1668
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1669
command = SetExtendedTimeoutCmd
1670
propname = "ExtendedTimeout"
1671
values_to_set = [datetime.timedelta(),
1672
datetime.timedelta(minutes=5),
1673
datetime.timedelta(seconds=1),
1674
datetime.timedelta(weeks=1),
1675
datetime.timedelta(weeks=52)]
1676
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1679
class TestSetIntervalCmd(TestPropertyValueCmd):
1680
command = SetIntervalCmd
1681
propname = "Interval"
1682
values_to_set = [datetime.timedelta(),
1683
datetime.timedelta(minutes=5),
1684
datetime.timedelta(seconds=1),
1685
datetime.timedelta(weeks=1),
1686
datetime.timedelta(weeks=52)]
1687
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1690
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1691
command = SetApprovalDelayCmd
1692
propname = "ApprovalDelay"
1693
values_to_set = [datetime.timedelta(),
1694
datetime.timedelta(minutes=5),
1695
datetime.timedelta(seconds=1),
1696
datetime.timedelta(weeks=1),
1697
datetime.timedelta(weeks=52)]
1698
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1701
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1702
command = SetApprovalDurationCmd
1703
propname = "ApprovalDuration"
1704
values_to_set = [datetime.timedelta(),
1705
datetime.timedelta(minutes=5),
1706
datetime.timedelta(seconds=1),
1707
datetime.timedelta(weeks=1),
1708
datetime.timedelta(weeks=52)]
1709
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1713
def should_only_run_tests():
1714
parser = argparse.ArgumentParser(add_help=False)
1715
parser.add_argument("--check", action='store_true')
1716
args, unknown_args = parser.parse_known_args()
1717
run_tests = args.check
1719
# Remove --check argument from sys.argv
1720
sys.argv[1:] = unknown_args
1723
# Add all tests from doctest strings
1724
def load_tests(loader, tests, none):
1726
tests.addTests(doctest.DocTestSuite())
337
1729
if __name__ == "__main__":
1731
if should_only_run_tests():
1732
# Call using ./tdd-python-script --check [--verbose]