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