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 953
- compare with another revision
- download diff
- download tarball
- view history from revision 953
- Committer: Teddy Hogeborn
- Date: 2018-08-19 01:35:11 UTC
- Revision ID: teddy@recompile.se-20180819013511-cku25q9yeub3dnr0
Adapt to changes in cryptsetup; use "cryptroot-unlock" program
* Makefile (install-client-nokey): Also install new script files
"mandos-to-cryptroot-unlock" and "initramfs-tools-script-stop".
* debian/mandos-client.dirs: Add
"usr/share/initramfs-tools/scripts/local-premount".
* initramfs-tools-hook: Also copy "mandos-to-cryptroot-unlock".
* initramfs-tools-script: Only modify keyscript setting in cryptroot
file if the file exists, otherwise start
"mandos-to-cryptroot-unlock" in background.
* initramfs-tools-script-stop: New script to make sure plugin-runner
has stopped before continuing.
* mandos-to-cryptroot-unlock: New script to run plugin-runner and feed
any password it gets into the "cryptroot-unlock" program.
* Makefile (install-client-nokey): Also install new script files
"mandos-to-cryptroot-unlock" and "initramfs-tools-script-stop".
* debian/mandos-client.dirs: Add
"usr/share/initramfs-tools/scripts/local-premount".
* initramfs-tools-hook: Also copy "mandos-to-cryptroot-unlock".
* initramfs-tools-script: Only modify keyscript setting in cryptroot
file if the file exists, otherwise start
"mandos-to-cryptroot-unlock" in background.
* initramfs-tools-script-stop: New script to make sure plugin-runner
has stopped before continuing.
* mandos-to-cryptroot-unlock: New script to run plugin-runner and feed
any password it gets into the "cryptroot-unlock" program.
- files added:
- initramfs-tools-hook-conf
- files removed:
- debian/mandos-client.templates
- files modified:
- DBUS-API
- initramfs-tools-script-stop *
- mandos-to-cryptroot-unlock *
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.19"
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
class TableOfClients(object):
272
tablewords = {
273
"Name": "Name",
274
"Enabled": "Enabled",
275
"Timeout": "Timeout",
276
"LastCheckedOK": "Last Successful Check",
277
"LastApprovalRequest": "Last Approval Request",
278
"Created": "Created",
279
"Interval": "Interval",
280
"Host": "Host",
281
"Fingerprint": "Fingerprint",
282
"KeyID": "Key ID",
283
"CheckerRunning": "Check Is Running",
284
"LastEnabled": "Last Enabled",
285
"ApprovalPending": "Approval Is Pending",
286
"ApprovedByDefault": "Approved By Default",
287
"ApprovalDelay": "Approval Delay",
288
"ApprovalDuration": "Approval Duration",
289
"Checker": "Checker",
290
"ExtendedTimeout": "Extended Timeout",
291
"Expires": "Expires",
292
"LastCheckerStatus": "Last Checker Status",
293
}
294
295
def __init__(self, clients, keywords, tablewords=None):
296
self.clients = clients
297
self.keywords = keywords
298
if tablewords is not None:
299
self.tablewords = tablewords
300
301
def __str__(self):
302
return "\n".join(self.rows())
303
304
if sys.version_info.major == 2:
305
__unicode__ = __str__
306
def __str__(self):
307
return str(self).encode(locale.getpreferredencoding())
308
309
def rows(self):
310
format_string = self.row_formatting_string()
311
rows = [self.header_line(format_string)]
312
rows.extend(self.client_line(client, format_string)
313
for client in self.clients)
314
return rows
315
316
def row_formatting_string(self):
317
"Format string used to format table rows"
318
return " ".join("{{{key}:{width}}}".format(
319
width=max(len(self.tablewords[key]),
320
*(len(self.string_from_client(client, key))
321
for client in self.clients)),
322
key=key)
323
for key in self.keywords)
324
325
def string_from_client(self, client, key):
326
return self.valuetostring(client[key], key)
327
328
@staticmethod
236
def print_clients(clients, keywords):
329
237
def valuetostring(value, keyword):
330
if isinstance(value, dbus.Boolean):
238
if type(value) is dbus.Boolean:
331
239
return "Yes" if value else "No"
332
240
if keyword in ("Timeout", "Interval", "ApprovalDelay",
333
241
"ApprovalDuration", "ExtendedTimeout"):
334
242
return milliseconds_to_string(value)
335
243
return str(value)
336
244
337
def header_line(self, format_string):
338
return format_string.format(**self.tablewords)
339
340
def client_line(self, client, format_string):
341
return format_string.format(
342
**{key: self.string_from_client(client, key)
343
for key in self.keywords})
344
345
346
## Classes for commands.
347
348
# Abstract classes first
349
class Command(object):
350
"""Abstract class for commands"""
351
def run(self, clients):
352
"""Normal commands should implement run_on_one_client(), but
353
commands which want to operate on all clients at the same time
354
can override this run() method instead."""
355
for client in clients:
356
self.run_on_one_client(client)
357
358
class PrintCmd(Command):
359
"""Abstract class for commands printing client details"""
360
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
361
"Created", "Interval", "Host", "KeyID",
362
"Fingerprint", "CheckerRunning", "LastEnabled",
363
"ApprovalPending", "ApprovedByDefault",
364
"LastApprovalRequest", "ApprovalDelay",
365
"ApprovalDuration", "Checker", "ExtendedTimeout",
366
"Expires", "LastCheckerStatus")
367
def run(self, clients):
368
print(self.output(clients))
369
370
class PropertyCmd(Command):
371
"""Abstract class for Actions for setting one client property"""
372
def run_on_one_client(self, client):
373
"""Set the Client's D-Bus property"""
374
client.Set(client_interface, self.property, self.value_to_set,
375
dbus_interface=dbus.PROPERTIES_IFACE)
376
377
class ValueArgumentMixIn(object):
378
"""Mixin class for commands taking a value as argument"""
379
def __init__(self, value):
380
self.value_to_set = value
381
382
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
383
"""Mixin class for commands taking a value argument as
384
milliseconds."""
385
@property
386
def value_to_set(self):
387
return self._vts
388
@value_to_set.setter
389
def value_to_set(self, value):
390
"""When setting, convert value to a datetime.timedelta"""
391
self._vts = string_to_delta(value).total_seconds() * 1000
392
393
# Actual (non-abstract) command classes
394
395
class PrintTableCmd(PrintCmd):
396
def __init__(self, verbose=False):
397
self.verbose = verbose
398
def output(self, clients):
399
if self.verbose:
400
keywords = self.all_keywords
401
else:
402
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
403
return str(TableOfClients(clients.values(), keywords))
404
405
class DumpJSONCmd(PrintCmd):
406
def output(self, clients):
407
data = {client["Name"]:
408
{key: self.dbus_boolean_to_bool(client[key])
409
for key in self.all_keywords}
410
for client in clients.values()}
411
return json.dumps(data, indent=4, separators=(',', ': '))
412
@staticmethod
413
def dbus_boolean_to_bool(value):
414
if isinstance(value, dbus.Boolean):
415
value = bool(value)
416
return value
417
418
class IsEnabledCmd(Command):
419
def run_on_one_client(self, client):
420
if self.is_enabled(client):
421
sys.exit(0)
422
sys.exit(1)
423
def is_enabled(self, client):
424
return client.Get(client_interface, "Enabled",
425
dbus_interface=dbus.PROPERTIES_IFACE)
426
427
class RemoveCmd(Command):
428
def __init__(self, mandos):
429
self.mandos = mandos
430
def run_on_one_client(self, client):
431
self.mandos.RemoveClient(client.__dbus_object_path__)
432
433
class ApproveCmd(Command):
434
def run_on_one_client(self, client):
435
client.Approve(dbus.Boolean(True),
436
dbus_interface=client_interface)
437
438
class DenyCmd(Command):
439
def run_on_one_client(self, client):
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"
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,
527
289
help="Print all fields")
528
290
parser.add_argument("-j", "--dump-json", action="store_true",
529
291
help="Dump client data in JSON format")
530
enable_disable = parser.add_mutually_exclusive_group()
531
enable_disable.add_argument("-e", "--enable", action="store_true",
532
help="Enable client")
533
enable_disable.add_argument("-d", "--disable",
534
action="store_true",
535
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")
536
296
parser.add_argument("-b", "--bump-timeout", action="store_true",
537
297
help="Bump timeout for client")
538
start_stop_checker = parser.add_mutually_exclusive_group()
539
start_stop_checker.add_argument("--start-checker",
540
action="store_true",
541
help="Start checker for client")
542
start_stop_checker.add_argument("--stop-checker",
543
action="store_true",
544
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")
545
302
parser.add_argument("-V", "--is-enabled", action="store_true",
546
303
help="Check if client is enabled")
547
304
parser.add_argument("-r", "--remove", action="store_true",
554
311
help="Set extended timeout for client")
555
312
parser.add_argument("-i", "--interval",
556
313
help="Set checker interval for client")
557
approve_deny_default = parser.add_mutually_exclusive_group()
558
approve_deny_default.add_argument(
559
"--approve-by-default", action="store_true",
560
default=None, dest="approved_by_default",
561
help="Set client to be approved by default")
562
approve_deny_default.add_argument(
563
"--deny-by-default", action="store_false",
564
dest="approved_by_default",
565
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")
566
320
parser.add_argument("--approval-delay",
567
321
help="Set delay before client approve/deny")
568
322
parser.add_argument("--approval-duration",
571
325
parser.add_argument("-s", "--secret",
572
326
type=argparse.FileType(mode="rb"),
573
327
help="Set password blob (file) for client")
574
approve_deny = parser.add_mutually_exclusive_group()
575
approve_deny.add_argument(
576
"-A", "--approve", action="store_true",
577
help="Approve any current client request")
578
approve_deny.add_argument("-D", "--deny", action="store_true",
579
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")
580
332
parser.add_argument("--check", action="store_true",
581
333
help="Run self-test")
582
334
parser.add_argument("client", nargs="*", help="Client name")
591
343
parser.error("--dump-json can only be used alone.")
592
344
if options.all and not has_actions(options):
593
345
parser.error("--all requires an action.")
594
if options.is_enabled and len(options.client) > 1:
595
parser.error("--is-enabled requires exactly one client")
346
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)
596
351
597
352
try:
598
353
bus = dbus.SystemBus()
599
354
mandos_dbus_objc = bus.get_object(busname, server_path)
600
355
except dbus.exceptions.DBusException:
601
log.critical("Could not connect to Mandos server")
356
print("Could not connect to Mandos server", file=sys.stderr)
602
357
sys.exit(1)
603
358
604
359
mandos_serv = dbus.Interface(mandos_dbus_objc,
606
361
mandos_serv_object_manager = dbus.Interface(
607
362
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
608
363
609
commands = []
610
611
if options.dump_json:
612
commands.append(DumpJSONCmd())
613
614
if options.enable:
615
commands.append(EnableCmd())
616
617
if options.disable:
618
commands.append(DisableCmd())
619
620
if options.bump_timeout:
621
commands.append(BumpTimeoutCmd(options.bump_timeout))
622
623
if options.start_checker:
624
commands.append(StartCheckerCmd())
625
626
if options.stop_checker:
627
commands.append(StopCheckerCmd())
628
629
if options.is_enabled:
630
commands.append(IsEnabledCmd())
631
632
if options.remove:
633
commands.append(RemoveCmd(mandos_serv))
634
635
if options.checker is not None:
636
commands.append(SetCheckerCmd())
637
638
if options.timeout is not None:
639
commands.append(SetTimeoutCmd(options.timeout))
640
641
if options.extended_timeout:
642
commands.append(
643
SetExtendedTimeoutCmd(options.extended_timeout))
644
645
if options.interval is not None:
646
command.append(SetIntervalCmd(options.interval))
647
648
if options.approved_by_default is not None:
649
if options.approved_by_default:
650
command.append(ApproveByDefaultCmd())
651
else:
652
command.append(DenyByDefaultCmd())
653
654
if options.approval_delay is not None:
655
command.append(SetApprovalDelayCmd(options.approval_delay))
656
657
if options.approval_duration is not None:
658
command.append(
659
SetApprovalDurationCmd(options.approval_duration))
660
661
if options.host is not None:
662
command.append(SetHostCmd(options.host))
663
664
if options.secret is not None:
665
command.append(SetSecretCmd(options.secret))
666
667
if options.approve:
668
commands.append(ApproveCmd())
669
670
if options.deny:
671
commands.append(DenyCmd())
672
673
# If no command option has been given, show table of clients,
674
# optionally verbosely
675
if not commands:
676
commands.append(PrintTableCmd(verbose=options.verbose))
677
678
364
# block stderr since dbus library prints to stderr
679
365
null = os.open(os.path.devnull, os.O_RDWR)
680
366
stderrcopy = os.dup(sys.stderr.fileno())
692
378
os.dup2(stderrcopy, sys.stderr.fileno())
693
379
os.close(stderrcopy)
694
380
except dbus.exceptions.DBusException as e:
695
log.critical("Failed to access Mandos server through D-Bus:"
696
"\n%s", e)
381
print("Access denied: "
382
"Accessing mandos server through D-Bus: {}".format(e),
383
file=sys.stderr)
697
384
sys.exit(1)
698
385
699
386
# Compile dict of (clients: properties) to process
710
397
clients[client_objc] = client
711
398
break
712
399
else:
713
log.critical("Client not found on server: %r", name)
400
print("Client not found on server: {!r}"
401
.format(name), file=sys.stderr)
714
402
sys.exit(1)
715
403
716
# Run all commands on clients
717
for command in commands:
718
command.run(clients)
719
720
721
class Test_milliseconds_to_string(unittest.TestCase):
722
def test_all(self):
723
self.assertEqual(milliseconds_to_string(93785000),
724
"1T02:03:05")
725
def test_no_days(self):
726
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
727
def test_all_zero(self):
728
self.assertEqual(milliseconds_to_string(0), "00:00:00")
729
def test_no_fractional_seconds(self):
730
self.assertEqual(milliseconds_to_string(400), "00:00:00")
731
self.assertEqual(milliseconds_to_string(900), "00:00:00")
732
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
733
734
class Test_string_to_delta(unittest.TestCase):
735
def test_handles_basic_rfc3339(self):
736
self.assertEqual(string_to_delta("PT2H"),
737
datetime.timedelta(0, 7200))
738
def test_falls_back_to_pre_1_6_1_with_warning(self):
739
# assertLogs only exists in Python 3.4
740
if hasattr(self, "assertLogs"):
741
with self.assertLogs(log, logging.WARNING):
742
value = string_to_delta("2h")
743
else:
744
value = string_to_delta("2h")
745
self.assertEqual(value, datetime.timedelta(0, 7200))
746
747
class Test_TableOfClients(unittest.TestCase):
748
def setUp(self):
749
self.tablewords = {
750
"Attr1": "X",
751
"AttrTwo": "Yy",
752
"AttrThree": "Zzz",
753
"Bool": "A D-BUS Boolean",
754
"NonDbusBoolean": "A Non-D-BUS Boolean",
755
"Integer": "An Integer",
756
"Timeout": "Timedelta 1",
757
"Interval": "Timedelta 2",
758
"ApprovalDelay": "Timedelta 3",
759
"ApprovalDuration": "Timedelta 4",
760
"ExtendedTimeout": "Timedelta 5",
761
"String": "A String",
762
}
763
self.keywords = ["Attr1", "AttrTwo"]
764
self.clients = [
765
{
766
"Attr1": "x1",
767
"AttrTwo": "y1",
768
"AttrThree": "z1",
769
"Bool": dbus.Boolean(False),
770
"NonDbusBoolean": False,
771
"Integer": 0,
772
"Timeout": 0,
773
"Interval": 1000,
774
"ApprovalDelay": 2000,
775
"ApprovalDuration": 3000,
776
"ExtendedTimeout": 4000,
777
"String": "",
778
},
779
{
780
"Attr1": "x2",
781
"AttrTwo": "y2",
782
"AttrThree": "z2",
783
"Bool": dbus.Boolean(True),
784
"NonDbusBoolean": True,
785
"Integer": 1,
786
"Timeout": 93785000,
787
"Interval": 93786000,
788
"ApprovalDelay": 93787000,
789
"ApprovalDuration": 93788000,
790
"ExtendedTimeout": 93789000,
791
"String": "A huge string which will not fit," * 10,
792
},
793
]
794
def test_short_header(self):
795
text = str(TableOfClients(self.clients, self.keywords,
796
self.tablewords))
797
expected_text = """
798
X Yy
799
x1 y1
800
x2 y2
801
"""[1:-1]
802
self.assertEqual(text, expected_text)
803
def test_booleans(self):
804
keywords = ["Bool", "NonDbusBoolean"]
805
text = str(TableOfClients(self.clients, keywords,
806
self.tablewords))
807
expected_text = """
808
A D-BUS Boolean A Non-D-BUS Boolean
809
No False
810
Yes True
811
"""[1:-1]
812
self.assertEqual(text, expected_text)
813
def test_milliseconds_detection(self):
814
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
815
"ApprovalDuration", "ExtendedTimeout"]
816
text = str(TableOfClients(self.clients, keywords,
817
self.tablewords))
818
expected_text = """
819
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
820
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
821
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
822
"""[1:-1]
823
self.assertEqual(text, expected_text)
824
def test_empty_and_long_string_values(self):
825
keywords = ["String"]
826
text = str(TableOfClients(self.clients, keywords,
827
self.tablewords))
828
expected_text = """
829
A String
830
831
A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,
832
"""[1:-1]
833
self.assertEqual(text, expected_text)
834
835
836
837
def should_only_run_tests():
838
parser = argparse.ArgumentParser(add_help=False)
839
parser.add_argument("--check", action='store_true')
840
args, unknown_args = parser.parse_known_args()
841
run_tests = args.check
842
if run_tests:
843
# Remove --check argument from sys.argv
844
sys.argv[1:] = unknown_args
845
return run_tests
846
847
# Add all tests from doctest strings
848
def load_tests(loader, tests, none):
849
import doctest
850
tests.addTests(doctest.DocTestSuite())
851
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
852
495
853
496
if __name__ == "__main__":
854
if should_only_run_tests():
855
# Call using ./tdd-python-script --check [--verbose]
856
unittest.main()
857
else:
858
main()
497
main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0