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