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 553
- compare with another revision
- download diff
- download tarball
- view history from revision 553
- Committer: Teddy Hogeborn
- Date: 2012-01-08 07:52:26 UTC
- Revision ID: teddy@recompile.se-20120108075226-ctx469ng75x5yfdd
Remove se.recompile.Mandos.NewRequest signal - its implemetation was
buggy and its utility was questionable.
* DBUS-API (se.recompile.Mandos.NewRequest): Removed.
* mandos (ClientDBus.NewRequest): Removed. All emitters altered.
buggy and its utility was questionable.
* DBUS-API (se.recompile.Mandos.NewRequest): Removed.
* mandos (ClientDBus.NewRequest): Removed. All emitters altered.
- 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-2012 Teddy Hogeborn
7
# Copyright © 2008-2012 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
47
import dbus
48
49
# Show warnings by default
50
if not sys.warnoptions:
51
import warnings
52
warnings.simplefilter("default")
53
54
log = logging.getLogger(sys.argv[0])
55
logging.basicConfig(level="INFO", # Show info level messages
56
format="%(message)s") # Show basic log messages
57
58
logging.captureWarnings(True) # Show warnings via the logging system
59
60
if sys.version_info.major == 2:
61
str = unicode
62
35
63
36
locale.setlocale(locale.LC_ALL, "")
64
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")
65
58
domain = "se.recompile"
66
59
busname = domain + ".Mandos"
67
60
server_path = "/"
68
61
server_interface = domain + ".Mandos"
69
62
client_interface = domain + ".Mandos.Client"
70
version = "1.8.3"
71
72
73
try:
74
dbus.OBJECT_MANAGER_IFACE
75
except AttributeError:
76
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
77
63
version = "1.5.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))
78
70
79
71
def milliseconds_to_string(ms):
80
72
td = datetime.timedelta(0, 0, 0, ms)
81
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
82
.format(days="{}T".format(td.days) if td.days else "",
83
hours=td.seconds // 3600,
84
minutes=(td.seconds % 3600) // 60,
85
seconds=td.seconds % 60))
86
87
88
def rfc3339_duration_to_delta(duration):
89
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
90
91
>>> rfc3339_duration_to_delta("P7D")
92
datetime.timedelta(7)
93
>>> rfc3339_duration_to_delta("PT60S")
94
datetime.timedelta(0, 60)
95
>>> rfc3339_duration_to_delta("PT60M")
96
datetime.timedelta(0, 3600)
97
>>> rfc3339_duration_to_delta("P60M")
98
datetime.timedelta(1680)
99
>>> rfc3339_duration_to_delta("PT24H")
100
datetime.timedelta(1)
101
>>> rfc3339_duration_to_delta("P1W")
102
datetime.timedelta(7)
103
>>> rfc3339_duration_to_delta("PT5M30S")
104
datetime.timedelta(0, 330)
105
>>> rfc3339_duration_to_delta("P1DT3M20S")
106
datetime.timedelta(1, 200)
107
>>> # Can not be empty:
108
>>> rfc3339_duration_to_delta("")
109
Traceback (most recent call last):
110
...
111
ValueError: Invalid RFC 3339 duration: u''
112
>>> # Must start with "P":
113
>>> rfc3339_duration_to_delta("1D")
114
Traceback (most recent call last):
115
...
116
ValueError: Invalid RFC 3339 duration: u'1D'
117
>>> # Must use correct order
118
>>> rfc3339_duration_to_delta("PT1S2M")
119
Traceback (most recent call last):
120
...
121
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
122
>>> # Time needs time marker
123
>>> rfc3339_duration_to_delta("P1H2S")
124
Traceback (most recent call last):
125
...
126
ValueError: Invalid RFC 3339 duration: u'P1H2S'
127
>>> # Weeks can not be combined with anything else
128
>>> rfc3339_duration_to_delta("P1D2W")
129
Traceback (most recent call last):
130
...
131
ValueError: Invalid RFC 3339 duration: u'P1D2W'
132
>>> rfc3339_duration_to_delta("P2W2H")
133
Traceback (most recent call last):
134
...
135
ValueError: Invalid RFC 3339 duration: u'P2W2H'
136
"""
137
138
# Parsing an RFC 3339 duration with regular expressions is not
139
# possible - there would have to be multiple places for the same
140
# values, like seconds. The current code, while more esoteric, is
141
# cleaner without depending on a parsing library. If Python had a
142
# built-in library for parsing we would use it, but we'd like to
143
# avoid excessive use of external libraries.
144
145
# New type for defining tokens, syntax, and semantics all-in-one
146
Token = collections.namedtuple("Token", (
147
"regexp", # To match token; if "value" is not None, must have
148
# a "group" containing digits
149
"value", # datetime.timedelta or None
150
"followers")) # Tokens valid after this token
151
# RFC 3339 "duration" tokens, syntax, and semantics; taken from
152
# the "duration" ABNF definition in RFC 3339, Appendix A.
153
token_end = Token(re.compile(r"$"), None, frozenset())
154
token_second = Token(re.compile(r"(\d+)S"),
155
datetime.timedelta(seconds=1),
156
frozenset((token_end, )))
157
token_minute = Token(re.compile(r"(\d+)M"),
158
datetime.timedelta(minutes=1),
159
frozenset((token_second, token_end)))
160
token_hour = Token(re.compile(r"(\d+)H"),
161
datetime.timedelta(hours=1),
162
frozenset((token_minute, token_end)))
163
token_time = Token(re.compile(r"T"),
164
None,
165
frozenset((token_hour, token_minute,
166
token_second)))
167
token_day = Token(re.compile(r"(\d+)D"),
168
datetime.timedelta(days=1),
169
frozenset((token_time, token_end)))
170
token_month = Token(re.compile(r"(\d+)M"),
171
datetime.timedelta(weeks=4),
172
frozenset((token_day, token_end)))
173
token_year = Token(re.compile(r"(\d+)Y"),
174
datetime.timedelta(weeks=52),
175
frozenset((token_month, token_end)))
176
token_week = Token(re.compile(r"(\d+)W"),
177
datetime.timedelta(weeks=1),
178
frozenset((token_end, )))
179
token_duration = Token(re.compile(r"P"), None,
180
frozenset((token_year, token_month,
181
token_day, token_time,
182
token_week)))
183
# Define starting values:
184
# Value so far
185
value = datetime.timedelta()
186
found_token = None
187
# Following valid tokens
188
followers = frozenset((token_duration, ))
189
# String left to parse
190
s = duration
191
# Loop until end token is found
192
while found_token is not token_end:
193
# Search for any currently valid tokens
194
for token in followers:
195
match = token.regexp.match(s)
196
if match is not None:
197
# Token found
198
if token.value is not None:
199
# Value found, parse digits
200
factor = int(match.group(1), 10)
201
# Add to value so far
202
value += factor * token.value
203
# Strip token from string
204
s = token.regexp.sub("", s, 1)
205
# Go to found token
206
found_token = token
207
# Set valid next tokens
208
followers = found_token.followers
209
break
210
else:
211
# No currently valid tokens were found
212
raise ValueError("Invalid RFC 3339 duration: {!r}"
213
.format(duration))
214
# End token found
215
return value
216
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
})
217
79
218
80
def string_to_delta(interval):
219
"""Parse a string and return a datetime.timedelta"""
220
221
try:
222
return rfc3339_duration_to_delta(interval)
223
except ValueError as e:
224
log.warning("%s - Parsing as pre-1.6.1 interval instead",
225
' '.join(e.args))
226
return parse_pre_1_6_1_interval(interval)
227
228
229
def parse_pre_1_6_1_interval(interval):
230
"""Parse an interval string as documented by Mandos before 1.6.1,
231
and return a datetime.timedelta
232
233
>>> parse_pre_1_6_1_interval('7d')
81
"""Parse a string and return a datetime.timedelta
82
83
>>> string_to_delta("7d")
234
84
datetime.timedelta(7)
235
>>> parse_pre_1_6_1_interval('60s')
85
>>> string_to_delta("60s")
236
86
datetime.timedelta(0, 60)
237
>>> parse_pre_1_6_1_interval('60m')
87
>>> string_to_delta("60m")
238
88
datetime.timedelta(0, 3600)
239
>>> parse_pre_1_6_1_interval('24h')
89
>>> string_to_delta("24h")
240
90
datetime.timedelta(1)
241
>>> parse_pre_1_6_1_interval('1w')
91
>>> string_to_delta("1w")
242
92
datetime.timedelta(7)
243
>>> parse_pre_1_6_1_interval('5m 30s')
93
>>> string_to_delta("5m 30s")
244
94
datetime.timedelta(0, 330)
245
>>> parse_pre_1_6_1_interval('')
246
datetime.timedelta(0)
247
>>> # Ignore unknown characters, allow any order and repetitions
248
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
249
datetime.timedelta(2, 480, 18000)
250
251
95
"""
252
253
96
value = datetime.timedelta(0)
254
regexp = re.compile(r"(\d+)([dsmhw]?)")
255
97
regexp = re.compile("(\d+)([dsmhw]?)")
98
256
99
for num, suffix in regexp.findall(interval):
257
100
if suffix == "d":
258
101
value += datetime.timedelta(int(num))
268
111
value += datetime.timedelta(0, 0, 0, int(num))
269
112
return value
270
113
271
272
## Classes for commands.
273
274
# Abstract classes first
275
class Command(object):
276
"""Abstract class for commands"""
277
def run(self, mandos, clients):
278
"""Normal commands should implement run_on_one_client(), but
279
commands which want to operate on all clients at the same time
280
can override this run() method instead."""
281
self.mandos = mandos
282
for client, properties in clients.items():
283
self.run_on_one_client(client, properties)
284
285
class PrintCmd(Command):
286
"""Abstract class for commands printing client details"""
287
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
288
"Created", "Interval", "Host", "KeyID",
289
"Fingerprint", "CheckerRunning", "LastEnabled",
290
"ApprovalPending", "ApprovedByDefault",
291
"LastApprovalRequest", "ApprovalDelay",
292
"ApprovalDuration", "Checker", "ExtendedTimeout",
293
"Expires", "LastCheckerStatus")
294
def run(self, mandos, clients):
295
print(self.output(clients))
296
297
class PropertyCmd(Command):
298
"""Abstract class for Actions for setting one client property"""
299
def run_on_one_client(self, client, properties):
300
"""Set the Client's D-Bus property"""
301
client.Set(client_interface, self.property, self.value_to_set,
302
dbus_interface=dbus.PROPERTIES_IFACE)
303
304
class ValueArgumentMixIn(object):
305
"""Mixin class for commands taking a value as argument"""
306
def __init__(self, value):
307
self.value_to_set = value
308
309
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
310
"""Mixin class for commands taking a value argument as
311
milliseconds."""
312
@property
313
def value_to_set(self):
314
return self._vts
315
@value_to_set.setter
316
def value_to_set(self, value):
317
"""When setting, convert value to a datetime.timedelta"""
318
self._vts = string_to_delta(value).total_seconds() * 1000
319
320
# Actual (non-abstract) command classes
321
322
class PrintTableCmd(PrintCmd):
323
def __init__(self, verbose=False):
324
self.verbose = verbose
325
326
def output(self, clients):
327
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
328
keywords = default_keywords
329
if self.verbose:
330
keywords = self.all_keywords
331
return str(self.TableOfClients(clients.values(), keywords))
332
333
class TableOfClients(object):
334
tableheaders = {
335
"Name": "Name",
336
"Enabled": "Enabled",
337
"Timeout": "Timeout",
338
"LastCheckedOK": "Last Successful Check",
339
"LastApprovalRequest": "Last Approval Request",
340
"Created": "Created",
341
"Interval": "Interval",
342
"Host": "Host",
343
"Fingerprint": "Fingerprint",
344
"KeyID": "Key ID",
345
"CheckerRunning": "Check Is Running",
346
"LastEnabled": "Last Enabled",
347
"ApprovalPending": "Approval Is Pending",
348
"ApprovedByDefault": "Approved By Default",
349
"ApprovalDelay": "Approval Delay",
350
"ApprovalDuration": "Approval Duration",
351
"Checker": "Checker",
352
"ExtendedTimeout": "Extended Timeout",
353
"Expires": "Expires",
354
"LastCheckerStatus": "Last Checker Status",
355
}
356
357
def __init__(self, clients, keywords, tableheaders=None):
358
self.clients = clients
359
self.keywords = keywords
360
if tableheaders is not None:
361
self.tableheaders = tableheaders
362
363
def __str__(self):
364
return "\n".join(self.rows())
365
366
if sys.version_info.major == 2:
367
__unicode__ = __str__
368
def __str__(self):
369
return str(self).encode(locale.getpreferredencoding())
370
371
def rows(self):
372
format_string = self.row_formatting_string()
373
rows = [self.header_line(format_string)]
374
rows.extend(self.client_line(client, format_string)
375
for client in self.clients)
376
return rows
377
378
def row_formatting_string(self):
379
"Format string used to format table rows"
380
return " ".join("{{{key}:{width}}}".format(
381
width=max(len(self.tableheaders[key]),
382
*(len(self.string_from_client(client, key))
383
for client in self.clients)),
384
key=key)
385
for key in self.keywords)
386
387
def string_from_client(self, client, key):
388
return self.valuetostring(client[key], key)
389
390
@staticmethod
391
def valuetostring(value, keyword):
392
if isinstance(value, dbus.Boolean):
393
return "Yes" if value else "No"
394
if keyword in ("Timeout", "Interval", "ApprovalDelay",
395
"ApprovalDuration", "ExtendedTimeout"):
396
return milliseconds_to_string(value)
397
return str(value)
398
399
def header_line(self, format_string):
400
return format_string.format(**self.tableheaders)
401
402
def client_line(self, client, format_string):
403
return format_string.format(
404
**{key: self.string_from_client(client, key)
405
for key in self.keywords})
406
407
408
409
class DumpJSONCmd(PrintCmd):
410
def output(self, clients):
411
data = {client["Name"]:
412
{key: self.dbus_boolean_to_bool(client[key])
413
for key in self.all_keywords}
414
for client in clients.values()}
415
return json.dumps(data, indent=4, separators=(',', ': '))
416
@staticmethod
417
def dbus_boolean_to_bool(value):
418
if isinstance(value, dbus.Boolean):
419
value = bool(value)
420
return value
421
422
class IsEnabledCmd(Command):
423
def run_on_one_client(self, client, properties):
424
if self.is_enabled(client, properties):
425
sys.exit(0)
426
sys.exit(1)
427
def is_enabled(self, client, properties):
428
return bool(properties["Enabled"])
429
430
class RemoveCmd(Command):
431
def run_on_one_client(self, client, properties):
432
self.mandos.RemoveClient(client.__dbus_object_path__)
433
434
class ApproveCmd(Command):
435
def run_on_one_client(self, client, properties):
436
client.Approve(dbus.Boolean(True),
437
dbus_interface=client_interface)
438
439
class DenyCmd(Command):
440
def run_on_one_client(self, client, properties):
441
client.Approve(dbus.Boolean(False),
442
dbus_interface=client_interface)
443
444
class EnableCmd(PropertyCmd):
445
property = "Enabled"
446
value_to_set = dbus.Boolean(True)
447
448
class DisableCmd(PropertyCmd):
449
property = "Enabled"
450
value_to_set = dbus.Boolean(False)
451
452
class BumpTimeoutCmd(PropertyCmd):
453
property = "LastCheckedOK"
454
value_to_set = ""
455
456
class StartCheckerCmd(PropertyCmd):
457
property = "CheckerRunning"
458
value_to_set = dbus.Boolean(True)
459
460
class StopCheckerCmd(PropertyCmd):
461
property = "CheckerRunning"
462
value_to_set = dbus.Boolean(False)
463
464
class ApproveByDefaultCmd(PropertyCmd):
465
property = "ApprovedByDefault"
466
value_to_set = dbus.Boolean(True)
467
468
class DenyByDefaultCmd(PropertyCmd):
469
property = "ApprovedByDefault"
470
value_to_set = dbus.Boolean(False)
471
472
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
473
property = "Checker"
474
475
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
476
property = "Host"
477
478
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
479
@property
480
def value_to_set(self):
481
return self._vts
482
@value_to_set.setter
483
def value_to_set(self, value):
484
"""When setting, read data from supplied file object"""
485
self._vts = value.read()
486
value.close()
487
property = "Secret"
488
489
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
490
property = "Timeout"
491
492
class SetExtendedTimeoutCmd(PropertyCmd,
493
MillisecondsValueArgumentMixIn):
494
property = "ExtendedTimeout"
495
496
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
497
property = "Interval"
498
499
class SetApprovalDelayCmd(PropertyCmd,
500
MillisecondsValueArgumentMixIn):
501
property = "ApprovalDelay"
502
503
class SetApprovalDurationCmd(PropertyCmd,
504
MillisecondsValueArgumentMixIn):
505
property = "ApprovalDuration"
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))
506
136
507
137
def has_actions(options):
508
138
return any((options.enable,
524
154
options.approve,
525
155
options.deny))
526
156
527
def add_command_line_options(parser):
157
def main():
158
parser = argparse.ArgumentParser()
528
159
parser.add_argument("--version", action="version",
529
version="%(prog)s {}".format(version),
160
version = "%%prog %s" % version,
530
161
help="show version number and exit")
531
162
parser.add_argument("-a", "--all", action="store_true",
532
163
help="Select all clients")
533
164
parser.add_argument("-v", "--verbose", action="store_true",
534
165
help="Print all fields")
535
parser.add_argument("-j", "--dump-json", action="store_true",
536
help="Dump client data in JSON format")
537
enable_disable = parser.add_mutually_exclusive_group()
538
enable_disable.add_argument("-e", "--enable", action="store_true",
539
help="Enable client")
540
enable_disable.add_argument("-d", "--disable",
541
action="store_true",
542
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")
543
170
parser.add_argument("-b", "--bump-timeout", action="store_true",
544
171
help="Bump timeout for client")
545
start_stop_checker = parser.add_mutually_exclusive_group()
546
start_stop_checker.add_argument("--start-checker",
547
action="store_true",
548
help="Start checker for client")
549
start_stop_checker.add_argument("--stop-checker",
550
action="store_true",
551
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")
552
176
parser.add_argument("-V", "--is-enabled", action="store_true",
553
177
help="Check if client is enabled")
554
178
parser.add_argument("-r", "--remove", action="store_true",
561
185
help="Set extended timeout for client")
562
186
parser.add_argument("-i", "--interval",
563
187
help="Set checker interval for client")
564
approve_deny_default = parser.add_mutually_exclusive_group()
565
approve_deny_default.add_argument(
566
"--approve-by-default", action="store_true",
567
default=None, dest="approved_by_default",
568
help="Set client to be approved by default")
569
approve_deny_default.add_argument(
570
"--deny-by-default", action="store_false",
571
dest="approved_by_default",
572
help="Set client to be denied by default")
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")
573
194
parser.add_argument("--approval-delay",
574
195
help="Set delay before client approve/deny")
575
196
parser.add_argument("--approval-duration",
576
197
help="Set duration of one client approval")
577
198
parser.add_argument("-H", "--host", help="Set host for client")
578
parser.add_argument("-s", "--secret",
579
type=argparse.FileType(mode="rb"),
199
parser.add_argument("-s", "--secret", type=file,
580
200
help="Set password blob (file) for client")
581
approve_deny = parser.add_mutually_exclusive_group()
582
approve_deny.add_argument(
583
"-A", "--approve", action="store_true",
584
help="Approve any current client request")
585
approve_deny.add_argument("-D", "--deny", action="store_true",
586
help="Deny any current client request")
587
parser.add_argument("--check", action="store_true",
588
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")
589
205
parser.add_argument("client", nargs="*", help="Client name")
590
591
592
def commands_from_options(options):
593
594
commands = []
595
596
if options.dump_json:
597
commands.append(DumpJSONCmd())
598
599
if options.enable:
600
commands.append(EnableCmd())
601
602
if options.disable:
603
commands.append(DisableCmd())
604
605
if options.bump_timeout:
606
commands.append(BumpTimeoutCmd())
607
608
if options.start_checker:
609
commands.append(StartCheckerCmd())
610
611
if options.stop_checker:
612
commands.append(StopCheckerCmd())
613
614
if options.is_enabled:
615
commands.append(IsEnabledCmd())
616
617
if options.remove:
618
commands.append(RemoveCmd())
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 no command option has been given, show table of clients,
659
# optionally verbosely
660
if not commands:
661
commands.append(PrintTableCmd(verbose=options.verbose))
662
663
return commands
664
665
666
def main():
667
parser = argparse.ArgumentParser()
668
669
add_command_line_options(parser)
670
671
206
options = parser.parse_args()
672
673
if has_actions(options) and not (options.client or options.all):
207
208
if has_actions(options) and not options.client and not options.all:
674
209
parser.error("Options require clients names or --all.")
675
210
if options.verbose and has_actions(options):
676
parser.error("--verbose can only be used alone.")
677
if options.dump_json and (options.verbose
678
or has_actions(options)):
679
parser.error("--dump-json can only be used alone.")
211
parser.error("--verbose can only be used alone or with"
212
" --all.")
680
213
if options.all and not has_actions(options):
681
214
parser.error("--all requires an action.")
682
if options.is_enabled and len(options.client) > 1:
683
parser.error("--is-enabled requires exactly one client")
684
685
clientnames = options.client
686
215
687
216
try:
688
217
bus = dbus.SystemBus()
689
218
mandos_dbus_objc = bus.get_object(busname, server_path)
690
219
except dbus.exceptions.DBusException:
691
log.critical("Could not connect to Mandos server")
220
print("Could not connect to Mandos server",
221
file=sys.stderr)
692
222
sys.exit(1)
693
223
694
224
mandos_serv = dbus.Interface(mandos_dbus_objc,
695
dbus_interface=server_interface)
696
mandos_serv_object_manager = dbus.Interface(
697
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
698
699
# Filter out log message from dbus module
700
dbus_logger = logging.getLogger("dbus.proxies")
701
class NullFilter(logging.Filter):
702
def filter(self, record):
703
return False
704
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)
705
232
try:
706
dbus_logger.addFilter(dbus_filter)
707
mandos_clients = {path: ifs_and_props[client_interface]
708
for path, ifs_and_props in
709
mandos_serv_object_manager
710
.GetManagedObjects().items()
711
if client_interface in ifs_and_props}
712
except dbus.exceptions.DBusException as e:
713
log.critical("Failed to access Mandos server through D-Bus:"
714
"\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)
715
242
sys.exit(1)
716
finally:
717
# restore dbus logger
718
dbus_logger.removeFilter(dbus_filter)
719
243
720
244
# Compile dict of (clients: properties) to process
721
clients = {}
722
723
if not clientnames:
724
clients = {bus.get_object(busname, path): properties
725
for path, 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())
726
251
else:
727
for name in clientnames:
728
for path, client in mandos_clients.items():
252
for name in options.client:
253
for path, client in mandos_clients.iteritems():
729
254
if client["Name"] == name:
730
255
client_objc = bus.get_object(busname, path)
731
256
clients[client_objc] = client
732
257
break
733
258
else:
734
log.critical("Client not found on server: %r", name)
259
print("Client not found on server: %r" % name,
260
file=sys.stderr)
735
261
sys.exit(1)
736
737
# Run all commands on clients
738
commands = commands_from_options(options)
739
for command in commands:
740
command.run(mandos_serv, clients)
741
742
743
class Test_milliseconds_to_string(unittest.TestCase):
744
def test_all(self):
745
self.assertEqual(milliseconds_to_string(93785000),
746
"1T02:03:05")
747
def test_no_days(self):
748
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
749
def test_all_zero(self):
750
self.assertEqual(milliseconds_to_string(0), "00:00:00")
751
def test_no_fractional_seconds(self):
752
self.assertEqual(milliseconds_to_string(400), "00:00:00")
753
self.assertEqual(milliseconds_to_string(900), "00:00:00")
754
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
755
756
class Test_string_to_delta(unittest.TestCase):
757
def test_handles_basic_rfc3339(self):
758
self.assertEqual(string_to_delta("PT0S"),
759
datetime.timedelta())
760
self.assertEqual(string_to_delta("P0D"),
761
datetime.timedelta())
762
self.assertEqual(string_to_delta("PT1S"),
763
datetime.timedelta(0, 1))
764
self.assertEqual(string_to_delta("PT2H"),
765
datetime.timedelta(0, 7200))
766
def test_falls_back_to_pre_1_6_1_with_warning(self):
767
# assertLogs only exists in Python 3.4
768
if hasattr(self, "assertLogs"):
769
with self.assertLogs(log, logging.WARNING):
770
value = string_to_delta("2h")
771
else:
772
class WarningFilter(logging.Filter):
773
"""Don't show, but record the presence of, warnings"""
774
def filter(self, record):
775
is_warning = record.levelno >= logging.WARNING
776
self.found = is_warning or getattr(self, "found",
777
False)
778
return not is_warning
779
warning_filter = WarningFilter()
780
log.addFilter(warning_filter)
781
try:
782
value = string_to_delta("2h")
783
finally:
784
log.removeFilter(warning_filter)
785
self.assertTrue(getattr(warning_filter, "found", False))
786
self.assertEqual(value, datetime.timedelta(0, 7200))
787
788
789
class TestCmd(unittest.TestCase):
790
"""Abstract class for tests of command classes"""
791
def setUp(self):
792
testcase = self
793
class MockClient(object):
794
def __init__(self, name, **attributes):
795
self.__dbus_object_path__ = "objpath_{}".format(name)
796
self.attributes = attributes
797
self.attributes["Name"] = name
798
self.calls = []
799
def Set(self, interface, property, value, dbus_interface):
800
testcase.assertEqual(interface, client_interface)
801
testcase.assertEqual(dbus_interface,
802
dbus.PROPERTIES_IFACE)
803
self.attributes[property] = value
804
def Get(self, interface, property, dbus_interface):
805
testcase.assertEqual(interface, client_interface)
806
testcase.assertEqual(dbus_interface,
807
dbus.PROPERTIES_IFACE)
808
return self.attributes[property]
809
def Approve(self, approve, dbus_interface):
810
testcase.assertEqual(dbus_interface, client_interface)
811
self.calls.append(("Approve", (approve,
812
dbus_interface)))
813
self.client = MockClient(
814
"foo",
815
KeyID=("92ed150794387c03ce684574b1139a65"
816
"94a34f895daaaf09fd8ea90a27cddb12"),
817
Secret=b"secret",
818
Host="foo.example.org",
819
Enabled=dbus.Boolean(True),
820
Timeout=300000,
821
LastCheckedOK="2019-02-03T00:00:00",
822
Created="2019-01-02T00:00:00",
823
Interval=120000,
824
Fingerprint=("778827225BA7DE539C5A"
825
"7CFA59CFF7CDBD9A5920"),
826
CheckerRunning=dbus.Boolean(False),
827
LastEnabled="2019-01-03T00:00:00",
828
ApprovalPending=dbus.Boolean(False),
829
ApprovedByDefault=dbus.Boolean(True),
830
LastApprovalRequest="",
831
ApprovalDelay=0,
832
ApprovalDuration=1000,
833
Checker="fping -q -- %(host)s",
834
ExtendedTimeout=900000,
835
Expires="2019-02-04T00:00:00",
836
LastCheckerStatus=0)
837
self.other_client = MockClient(
838
"barbar",
839
KeyID=("0558568eedd67d622f5c83b35a115f79"
840
"6ab612cff5ad227247e46c2b020f441c"),
841
Secret=b"secretbar",
842
Host="192.0.2.3",
843
Enabled=dbus.Boolean(True),
844
Timeout=300000,
845
LastCheckedOK="2019-02-04T00:00:00",
846
Created="2019-01-03T00:00:00",
847
Interval=120000,
848
Fingerprint=("3E393AEAEFB84C7E89E2"
849
"F547B3A107558FCA3A27"),
850
CheckerRunning=dbus.Boolean(True),
851
LastEnabled="2019-01-04T00:00:00",
852
ApprovalPending=dbus.Boolean(False),
853
ApprovedByDefault=dbus.Boolean(False),
854
LastApprovalRequest="2019-01-03T00:00:00",
855
ApprovalDelay=30000,
856
ApprovalDuration=1000,
857
Checker=":",
858
ExtendedTimeout=900000,
859
Expires="2019-02-05T00:00:00",
860
LastCheckerStatus=-2)
861
self.clients = collections.OrderedDict(
862
[
863
(self.client, self.client.attributes),
864
(self.other_client, self.other_client.attributes),
865
])
866
self.one_client = {self.client: self.client.attributes}
867
868
class TestPrintTableCmd(TestCmd):
869
def test_normal(self):
870
output = PrintTableCmd().output(self.clients)
871
expected_output = """
872
Name Enabled Timeout Last Successful Check
873
foo Yes 00:05:00 2019-02-03T00:00:00
874
barbar Yes 00:05:00 2019-02-04T00:00:00
875
"""[1:-1]
876
self.assertEqual(output, expected_output)
877
def test_verbose(self):
878
output = PrintTableCmd(verbose=True).output(self.clients)
879
expected_output = """
880
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
881
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
882
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
883
"""[1:-1]
884
self.assertEqual(output, expected_output)
885
def test_one_client(self):
886
output = PrintTableCmd().output(self.one_client)
887
expected_output = """
888
Name Enabled Timeout Last Successful Check
889
foo Yes 00:05:00 2019-02-03T00:00:00
890
"""[1:-1]
891
self.assertEqual(output, expected_output)
892
893
class TestDumpJSONCmd(TestCmd):
894
def setUp(self):
895
self.expected_json = {
896
"foo": {
897
"Name": "foo",
898
"KeyID": ("92ed150794387c03ce684574b1139a65"
899
"94a34f895daaaf09fd8ea90a27cddb12"),
900
"Host": "foo.example.org",
901
"Enabled": True,
902
"Timeout": 300000,
903
"LastCheckedOK": "2019-02-03T00:00:00",
904
"Created": "2019-01-02T00:00:00",
905
"Interval": 120000,
906
"Fingerprint": ("778827225BA7DE539C5A"
907
"7CFA59CFF7CDBD9A5920"),
908
"CheckerRunning": False,
909
"LastEnabled": "2019-01-03T00:00:00",
910
"ApprovalPending": False,
911
"ApprovedByDefault": True,
912
"LastApprovalRequest": "",
913
"ApprovalDelay": 0,
914
"ApprovalDuration": 1000,
915
"Checker": "fping -q -- %(host)s",
916
"ExtendedTimeout": 900000,
917
"Expires": "2019-02-04T00:00:00",
918
"LastCheckerStatus": 0,
919
},
920
"barbar": {
921
"Name": "barbar",
922
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
923
"6ab612cff5ad227247e46c2b020f441c"),
924
"Host": "192.0.2.3",
925
"Enabled": True,
926
"Timeout": 300000,
927
"LastCheckedOK": "2019-02-04T00:00:00",
928
"Created": "2019-01-03T00:00:00",
929
"Interval": 120000,
930
"Fingerprint": ("3E393AEAEFB84C7E89E2"
931
"F547B3A107558FCA3A27"),
932
"CheckerRunning": True,
933
"LastEnabled": "2019-01-04T00:00:00",
934
"ApprovalPending": False,
935
"ApprovedByDefault": False,
936
"LastApprovalRequest": "2019-01-03T00:00:00",
937
"ApprovalDelay": 30000,
938
"ApprovalDuration": 1000,
939
"Checker": ":",
940
"ExtendedTimeout": 900000,
941
"Expires": "2019-02-05T00:00:00",
942
"LastCheckerStatus": -2,
943
},
944
}
945
return super(TestDumpJSONCmd, self).setUp()
946
def test_normal(self):
947
json_data = json.loads(DumpJSONCmd().output(self.clients))
948
self.assertDictEqual(json_data, self.expected_json)
949
def test_one_client(self):
950
clients = self.one_client
951
json_data = json.loads(DumpJSONCmd().output(clients))
952
expected_json = {"foo": self.expected_json["foo"]}
953
self.assertDictEqual(json_data, expected_json)
954
955
class TestIsEnabledCmd(TestCmd):
956
def test_is_enabled(self):
957
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
958
for client, properties in self.clients.items()))
959
def test_is_enabled_run_exits_successfully(self):
960
with self.assertRaises(SystemExit) as e:
961
IsEnabledCmd().run(None, self.one_client)
962
if e.exception.code is not None:
963
self.assertEqual(e.exception.code, 0)
964
else:
965
self.assertIsNone(e.exception.code)
966
def test_is_enabled_run_exits_with_failure(self):
967
self.client.attributes["Enabled"] = dbus.Boolean(False)
968
with self.assertRaises(SystemExit) as e:
969
IsEnabledCmd().run(None, self.one_client)
970
if isinstance(e.exception.code, int):
971
self.assertNotEqual(e.exception.code, 0)
972
else:
973
self.assertIsNotNone(e.exception.code)
974
975
class TestRemoveCmd(TestCmd):
976
def test_remove(self):
977
class MockMandos(object):
978
def __init__(self):
979
self.calls = []
980
def RemoveClient(self, dbus_path):
981
self.calls.append(("RemoveClient", (dbus_path,)))
982
mandos = MockMandos()
983
super(TestRemoveCmd, self).setUp()
984
RemoveCmd().run(mandos, self.clients)
985
self.assertEqual(len(mandos.calls), 2)
986
for client in self.clients:
987
self.assertIn(("RemoveClient",
988
(client.__dbus_object_path__,)),
989
mandos.calls)
990
991
class TestApproveCmd(TestCmd):
992
def test_approve(self):
993
ApproveCmd().run(None, self.clients)
994
for client in self.clients:
995
self.assertIn(("Approve", (True, client_interface)),
996
client.calls)
997
998
class TestDenyCmd(TestCmd):
999
def test_deny(self):
1000
DenyCmd().run(None, self.clients)
1001
for client in self.clients:
1002
self.assertIn(("Approve", (False, client_interface)),
1003
client.calls)
1004
1005
class TestEnableCmd(TestCmd):
1006
def test_enable(self):
1007
for client in self.clients:
1008
client.attributes["Enabled"] = False
1009
1010
EnableCmd().run(None, self.clients)
1011
1012
for client in self.clients:
1013
self.assertTrue(client.attributes["Enabled"])
1014
1015
class TestDisableCmd(TestCmd):
1016
def test_disable(self):
1017
DisableCmd().run(None, self.clients)
1018
1019
for client in self.clients:
1020
self.assertFalse(client.attributes["Enabled"])
1021
1022
class Unique(object):
1023
"""Class for objects which exist only to be unique objects, since
1024
unittest.mock.sentinel only exists in Python 3.3"""
1025
1026
class TestPropertyCmd(TestCmd):
1027
"""Abstract class for tests of PropertyCmd classes"""
1028
def runTest(self):
1029
if not hasattr(self, "command"):
1030
return
1031
values_to_get = getattr(self, "values_to_get",
1032
self.values_to_set)
1033
for value_to_set, value_to_get in zip(self.values_to_set,
1034
values_to_get):
1035
for client in self.clients:
1036
old_value = client.attributes[self.property]
1037
self.assertNotIsInstance(old_value, Unique)
1038
client.attributes[self.property] = Unique()
1039
self.run_command(value_to_set, self.clients)
1040
for client in self.clients:
1041
value = client.attributes[self.property]
1042
self.assertNotIsInstance(value, Unique)
1043
self.assertEqual(value, value_to_get)
1044
def run_command(self, value, clients):
1045
self.command().run(None, clients)
1046
1047
class TestBumpTimeoutCmd(TestPropertyCmd):
1048
command = BumpTimeoutCmd
1049
property = "LastCheckedOK"
1050
values_to_set = [""]
1051
1052
class TestStartCheckerCmd(TestPropertyCmd):
1053
command = StartCheckerCmd
1054
property = "CheckerRunning"
1055
values_to_set = [dbus.Boolean(True)]
1056
1057
class TestStopCheckerCmd(TestPropertyCmd):
1058
command = StopCheckerCmd
1059
property = "CheckerRunning"
1060
values_to_set = [dbus.Boolean(False)]
1061
1062
class TestApproveByDefaultCmd(TestPropertyCmd):
1063
command = ApproveByDefaultCmd
1064
property = "ApprovedByDefault"
1065
values_to_set = [dbus.Boolean(True)]
1066
1067
class TestDenyByDefaultCmd(TestPropertyCmd):
1068
command = DenyByDefaultCmd
1069
property = "ApprovedByDefault"
1070
values_to_set = [dbus.Boolean(False)]
1071
1072
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1073
"""Abstract class for tests of PropertyCmd classes using the
1074
ValueArgumentMixIn"""
1075
def runTest(self):
1076
if type(self) is TestValueArgumentPropertyCmd:
1077
return
1078
return super(TestValueArgumentPropertyCmd, self).runTest()
1079
def run_command(self, value, clients):
1080
self.command(value).run(None, clients)
1081
1082
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1083
command = SetCheckerCmd
1084
property = "Checker"
1085
values_to_set = ["", ":", "fping -q -- %s"]
1086
1087
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1088
command = SetHostCmd
1089
property = "Host"
1090
values_to_set = ["192.0.2.3", "foo.example.org"]
1091
1092
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1093
command = SetSecretCmd
1094
property = "Secret"
1095
values_to_set = [open("/dev/null", "rb"),
1096
io.BytesIO(b"secret\0xyzzy\nbar")]
1097
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1098
1099
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1100
command = SetTimeoutCmd
1101
property = "Timeout"
1102
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1103
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1104
1105
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1106
command = SetExtendedTimeoutCmd
1107
property = "ExtendedTimeout"
1108
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1109
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1110
1111
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1112
command = SetIntervalCmd
1113
property = "Interval"
1114
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1115
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1116
1117
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1118
command = SetApprovalDelayCmd
1119
property = "ApprovalDelay"
1120
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1121
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1122
1123
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1124
command = SetApprovalDurationCmd
1125
property = "ApprovalDuration"
1126
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1127
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1128
1129
class TestOptions(unittest.TestCase):
1130
def setUp(self):
1131
self.parser = argparse.ArgumentParser()
1132
add_command_line_options(self.parser)
1133
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1134
"""Assert that parsing ARGS should result in an instance of
1135
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1136
options = self.parser.parse_args(args)
1137
commands = commands_from_options(options)
1138
self.assertEqual(len(commands), 1)
1139
command = commands[0]
1140
self.assertIsInstance(command, command_cls)
1141
for key, value in cmd_attrs.items():
1142
self.assertEqual(getattr(command, key), value)
1143
def test_default_is_show_table(self):
1144
self.assert_command_from_args([], PrintTableCmd,
1145
verbose=False)
1146
def test_show_table_verbose(self):
1147
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1148
verbose=True)
1149
def test_enable(self):
1150
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1151
def test_disable(self):
1152
self.assert_command_from_args(["--disable", "foo"],
1153
DisableCmd)
1154
1155
1156
1157
def should_only_run_tests():
1158
parser = argparse.ArgumentParser(add_help=False)
1159
parser.add_argument("--check", action='store_true')
1160
args, unknown_args = parser.parse_known_args()
1161
run_tests = args.check
1162
if run_tests:
1163
# Remove --check argument from sys.argv
1164
sys.argv[1:] = unknown_args
1165
return run_tests
1166
1167
# Add all tests from doctest strings
1168
def load_tests(loader, tests, none):
1169
import doctest
1170
tests.addTests(doctest.DocTestSuite())
1171
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(options.secret.read()),
335
dbus_interface=dbus.PROPERTIES_IFACE)
336
if options.approved_by_default is not None:
337
client.Set(client_interface, "ApprovedByDefault",
338
dbus.Boolean(options
339
.approved_by_default),
340
dbus_interface=dbus.PROPERTIES_IFACE)
341
if options.approve:
342
client.Approve(dbus.Boolean(True),
343
dbus_interface=client_interface)
344
elif options.deny:
345
client.Approve(dbus.Boolean(False),
346
dbus_interface=client_interface)
1172
347
1173
348
if __name__ == "__main__":
1174
if should_only_run_tests():
1175
# Call using ./tdd-python-script --check [--verbose]
1176
unittest.main()
1177
else:
1178
main()
349
main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0