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 979
- compare with another revision
- download diff
- download tarball
- view history from revision 979
- Committer: Teddy Hogeborn
- Date: 2019-02-11 07:06:55 UTC
- Revision ID: teddy@recompile.se-20190211070655-gzg2zowi1cq98cof
Update copyright year to 2019
* plugins.d/askpass-fifo.c: Update copyright year to 2019.
* plugins.d/password-prompt.c: - '' -
* plugins.d/askpass-fifo.c: Update copyright year to 2019.
* plugins.d/password-prompt.c: - '' -
- files modified:
- TODO
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
#
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
"KeyID": "Key ID",
62
"CheckerRunning": "Check Is Running",
63
"LastEnabled": "Last Enabled",
64
"ApprovalPending": "Approval Is Pending",
65
"ApprovedByDefault": "Approved By Default",
66
"ApprovalDelay": "Approval Delay",
67
"ApprovalDuration": "Approval Duration",
68
"Checker": "Checker",
69
"ExtendedTimeout": "Extended Timeout",
70
"Expires": "Expires",
71
"LastCheckerStatus": "Last Checker Status",
72
}
64
73
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
65
74
domain = "se.recompile"
66
75
busname = domain + ".Mandos"
94
103
datetime.timedelta(0, 60)
95
104
>>> rfc3339_duration_to_delta("PT60M")
96
105
datetime.timedelta(0, 3600)
97
>>> rfc3339_duration_to_delta("P60M")
98
datetime.timedelta(1680)
99
106
>>> rfc3339_duration_to_delta("PT24H")
100
107
datetime.timedelta(1)
101
108
>>> rfc3339_duration_to_delta("P1W")
104
111
datetime.timedelta(0, 330)
105
112
>>> rfc3339_duration_to_delta("P1DT3M20S")
106
113
datetime.timedelta(1, 200)
107
>>> # Can not be empty:
108
>>> rfc3339_duration_to_delta("")
109
Traceback (most recent call last):
110
...
111
ValueError: Invalid RFC 3339 duration: u''
112
>>> # Must start with "P":
113
>>> rfc3339_duration_to_delta("1D")
114
Traceback (most recent call last):
115
...
116
ValueError: Invalid RFC 3339 duration: u'1D'
117
>>> # Must use correct order
118
>>> rfc3339_duration_to_delta("PT1S2M")
119
Traceback (most recent call last):
120
...
121
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
122
>>> # Time needs time marker
123
>>> rfc3339_duration_to_delta("P1H2S")
124
Traceback (most recent call last):
125
...
126
ValueError: Invalid RFC 3339 duration: u'P1H2S'
127
>>> # Weeks can not be combined with anything else
128
>>> rfc3339_duration_to_delta("P1D2W")
129
Traceback (most recent call last):
130
...
131
ValueError: Invalid RFC 3339 duration: u'P1D2W'
132
>>> rfc3339_duration_to_delta("P2W2H")
133
Traceback (most recent call last):
134
...
135
ValueError: Invalid RFC 3339 duration: u'P2W2H'
136
114
"""
137
115
138
116
# Parsing an RFC 3339 duration with regular expressions is not
216
194
217
195
218
196
def string_to_delta(interval):
219
"""Parse a string and return a datetime.timedelta"""
197
"""Parse a string and return a datetime.timedelta
198
199
>>> string_to_delta('7d')
200
datetime.timedelta(7)
201
>>> string_to_delta('60s')
202
datetime.timedelta(0, 60)
203
>>> string_to_delta('60m')
204
datetime.timedelta(0, 3600)
205
>>> string_to_delta('24h')
206
datetime.timedelta(1)
207
>>> string_to_delta('1w')
208
datetime.timedelta(7)
209
>>> string_to_delta('5m 30s')
210
datetime.timedelta(0, 330)
211
"""
220
212
221
213
try:
222
214
return rfc3339_duration_to_delta(interval)
223
except ValueError as e:
224
log.warning("%s - Parsing as pre-1.6.1 interval instead",
225
' '.join(e.args))
226
return parse_pre_1_6_1_interval(interval)
227
228
229
def parse_pre_1_6_1_interval(interval):
230
"""Parse an interval string as documented by Mandos before 1.6.1,
231
and return a datetime.timedelta
232
233
>>> parse_pre_1_6_1_interval('7d')
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
"""
215
except ValueError:
216
pass
252
217
253
218
value = datetime.timedelta(0)
254
219
regexp = re.compile(r"(\d+)([dsmhw]?)")
269
234
return value
270
235
271
236
272
class TableOfClients(object):
273
tablewords = {
274
"Name": "Name",
275
"Enabled": "Enabled",
276
"Timeout": "Timeout",
277
"LastCheckedOK": "Last Successful Check",
278
"LastApprovalRequest": "Last Approval Request",
279
"Created": "Created",
280
"Interval": "Interval",
281
"Host": "Host",
282
"Fingerprint": "Fingerprint",
283
"KeyID": "Key ID",
284
"CheckerRunning": "Check Is Running",
285
"LastEnabled": "Last Enabled",
286
"ApprovalPending": "Approval Is Pending",
287
"ApprovedByDefault": "Approved By Default",
288
"ApprovalDelay": "Approval Delay",
289
"ApprovalDuration": "Approval Duration",
290
"Checker": "Checker",
291
"ExtendedTimeout": "Extended Timeout",
292
"Expires": "Expires",
293
"LastCheckerStatus": "Last Checker Status",
294
}
295
296
def __init__(self, clients, keywords, tablewords=None):
297
self.clients = clients
298
self.keywords = keywords
299
if tablewords is not None:
300
self.tablewords = tablewords
301
302
def __str__(self):
303
return "\n".join(self.rows())
304
305
if sys.version_info.major == 2:
306
__unicode__ = __str__
307
def __str__(self):
308
return str(self).encode(locale.getpreferredencoding())
309
310
def rows(self):
311
format_string = self.row_formatting_string()
312
rows = [self.header_line(format_string)]
313
rows.extend(self.client_line(client, format_string)
314
for client in self.clients)
315
return rows
316
317
def row_formatting_string(self):
318
"Format string used to format table rows"
319
return " ".join("{{{key}:{width}}}".format(
320
width=max(len(self.tablewords[key]),
321
*(len(self.string_from_client(client, key))
322
for client in self.clients)),
323
key=key)
324
for key in self.keywords)
325
326
def string_from_client(self, client, key):
327
return self.valuetostring(client[key], key)
328
329
@staticmethod
237
def print_clients(clients, keywords):
330
238
def valuetostring(value, keyword):
331
if isinstance(value, dbus.Boolean):
239
if type(value) is dbus.Boolean:
332
240
return "Yes" if value else "No"
333
241
if keyword in ("Timeout", "Interval", "ApprovalDelay",
334
242
"ApprovalDuration", "ExtendedTimeout"):
335
243
return milliseconds_to_string(value)
336
244
return str(value)
337
245
338
def header_line(self, format_string):
339
return format_string.format(**self.tablewords)
340
341
def client_line(self, client, format_string):
342
return format_string.format(
343
**{key: self.string_from_client(client, key)
344
for key in self.keywords})
246
# Create format string to print table rows
247
format_string = " ".join("{{{key}:{width}}}".format(
248
width=max(len(tablewords[key]),
249
max(len(valuetostring(client[key], key))
250
for client in clients)),
251
key=key)
252
for key in keywords)
253
# Print header line
254
print(format_string.format(**tablewords))
255
for client in clients:
256
print(format_string
257
.format(**{key: valuetostring(client[key], key)
258
for key in keywords}))
345
259
346
260
347
261
def has_actions(options):
431
345
if options.all and not has_actions(options):
432
346
parser.error("--all requires an action.")
433
347
348
if options.check:
349
import doctest
350
fail_count, test_count = doctest.testmod()
351
sys.exit(os.EX_OK if fail_count == 0 else 1)
352
434
353
try:
435
354
bus = dbus.SystemBus()
436
355
mandos_dbus_objc = bus.get_object(busname, server_path)
437
356
except dbus.exceptions.DBusException:
438
log.critical("Could not connect to Mandos server")
357
print("Could not connect to Mandos server", file=sys.stderr)
439
358
sys.exit(1)
440
359
441
360
mandos_serv = dbus.Interface(mandos_dbus_objc,
460
379
os.dup2(stderrcopy, sys.stderr.fileno())
461
380
os.close(stderrcopy)
462
381
except dbus.exceptions.DBusException as e:
463
log.critical("Failed to access Mandos server through D-Bus:"
464
"\n%s", e)
382
print("Access denied: "
383
"Accessing mandos server through D-Bus: {}".format(e),
384
file=sys.stderr)
465
385
sys.exit(1)
466
386
467
387
# Compile dict of (clients: properties) to process
478
398
clients[client_objc] = client
479
399
break
480
400
else:
481
log.critical("Client not found on server: %r", name)
401
print("Client not found on server: {!r}"
402
.format(name), file=sys.stderr)
482
403
sys.exit(1)
483
404
484
405
if not has_actions(options) and clients:
506
427
separators=(',', ': '))
507
428
print()
508
429
else:
509
print(TableOfClients(clients.values(), keywords))
430
print_clients(clients.values(), keywords)
510
431
else:
511
432
# Process each client in the list by all selected options
512
433
for client in clients:
572
493
client.Approve(dbus.Boolean(False),
573
494
dbus_interface=client_interface)
574
495
575
576
class Test_milliseconds_to_string(unittest.TestCase):
577
def test_all(self):
578
self.assertEqual(milliseconds_to_string(93785000),
579
"1T02:03:05")
580
def test_no_days(self):
581
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
582
def test_all_zero(self):
583
self.assertEqual(milliseconds_to_string(0), "00:00:00")
584
def test_no_fractional_seconds(self):
585
self.assertEqual(milliseconds_to_string(400), "00:00:00")
586
self.assertEqual(milliseconds_to_string(900), "00:00:00")
587
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
588
589
class Test_string_to_delta(unittest.TestCase):
590
def test_handles_basic_rfc3339(self):
591
self.assertEqual(string_to_delta("PT2H"),
592
datetime.timedelta(0, 7200))
593
def test_falls_back_to_pre_1_6_1_with_warning(self):
594
# assertLogs only exists in Python 3.4
595
if hasattr(self, "assertLogs"):
596
with self.assertLogs(log, logging.WARNING):
597
value = string_to_delta("2h")
598
else:
599
value = string_to_delta("2h")
600
self.assertEqual(value, datetime.timedelta(0, 7200))
601
602
class Test_TableOfClients(unittest.TestCase):
603
def setUp(self):
604
self.tablewords = {
605
"Attr1": "X",
606
"AttrTwo": "Yy",
607
"AttrThree": "Zzz",
608
"Bool": "A D-BUS Boolean",
609
"NonDbusBoolean": "A Non-D-BUS Boolean",
610
"Integer": "An Integer",
611
"Timeout": "Timedelta 1",
612
"Interval": "Timedelta 2",
613
"ApprovalDelay": "Timedelta 3",
614
"ApprovalDuration": "Timedelta 4",
615
"ExtendedTimeout": "Timedelta 5",
616
"String": "A String",
617
}
618
self.keywords = ["Attr1", "AttrTwo"]
619
self.clients = [
620
{
621
"Attr1": "x1",
622
"AttrTwo": "y1",
623
"AttrThree": "z1",
624
"Bool": dbus.Boolean(False),
625
"NonDbusBoolean": False,
626
"Integer": 0,
627
"Timeout": 0,
628
"Interval": 1000,
629
"ApprovalDelay": 2000,
630
"ApprovalDuration": 3000,
631
"ExtendedTimeout": 4000,
632
"String": "",
633
},
634
{
635
"Attr1": "x2",
636
"AttrTwo": "y2",
637
"AttrThree": "z2",
638
"Bool": dbus.Boolean(True),
639
"NonDbusBoolean": True,
640
"Integer": 1,
641
"Timeout": 93785000,
642
"Interval": 93786000,
643
"ApprovalDelay": 93787000,
644
"ApprovalDuration": 93788000,
645
"ExtendedTimeout": 93789000,
646
"String": "A huge string which will not fit," * 10,
647
},
648
]
649
def test_short_header(self):
650
text = str(TableOfClients(self.clients, self.keywords,
651
self.tablewords))
652
expected_text = """
653
X Yy
654
x1 y1
655
x2 y2
656
"""[1:-1]
657
self.assertEqual(text, expected_text)
658
def test_booleans(self):
659
keywords = ["Bool", "NonDbusBoolean"]
660
text = str(TableOfClients(self.clients, keywords,
661
self.tablewords))
662
expected_text = """
663
A D-BUS Boolean A Non-D-BUS Boolean
664
No False
665
Yes True
666
"""[1:-1]
667
self.assertEqual(text, expected_text)
668
def test_milliseconds_detection(self):
669
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
670
"ApprovalDuration", "ExtendedTimeout"]
671
text = str(TableOfClients(self.clients, keywords,
672
self.tablewords))
673
expected_text = """
674
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
675
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
676
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
677
"""[1:-1]
678
self.assertEqual(text, expected_text)
679
def test_empty_and_long_string_values(self):
680
keywords = ["String"]
681
text = str(TableOfClients(self.clients, keywords,
682
self.tablewords))
683
expected_text = """
684
A String
685
686
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,
687
"""[1:-1]
688
self.assertEqual(text, expected_text)
689
690
691
692
def should_only_run_tests():
693
parser = argparse.ArgumentParser(add_help=False)
694
parser.add_argument("--check", action='store_true')
695
args, unknown_args = parser.parse_known_args()
696
run_tests = args.check
697
if run_tests:
698
# Remove --check argument from sys.argv
699
sys.argv[1:] = unknown_args
700
return run_tests
701
702
# Add all tests from doctest strings
703
def load_tests(loader, tests, none):
704
import doctest
705
tests.addTests(doctest.DocTestSuite())
706
return tests
707
496
708
497
if __name__ == "__main__":
709
if should_only_run_tests():
710
# Call using ./tdd-python-script --check [--verbose]
711
unittest.main()
712
else:
713
main()
498
main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0