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