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 974
- compare with another revision
- download diff
- download tarball
- view history from revision 974
- Committer: Teddy Hogeborn
- Date: 2019-02-10 10:58:42 UTC
- mfrom: (237.4.106 release)
- Revision ID: teddy@recompile.se-20190210105842-l7htgtxjr23p4heb
Merge from release branch
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
89
76
server_path = "/"
90
77
server_interface = domain + ".Mandos"
91
78
client_interface = domain + ".Mandos.Client"
92
version = "1.8.3"
79
version = "1.8.2"
93
80
94
81
95
82
try:
116
103
datetime.timedelta(0, 60)
117
104
>>> rfc3339_duration_to_delta("PT60M")
118
105
datetime.timedelta(0, 3600)
119
>>> rfc3339_duration_to_delta("P60M")
120
datetime.timedelta(1680)
121
106
>>> rfc3339_duration_to_delta("PT24H")
122
107
datetime.timedelta(1)
123
108
>>> rfc3339_duration_to_delta("P1W")
126
111
datetime.timedelta(0, 330)
127
112
>>> rfc3339_duration_to_delta("P1DT3M20S")
128
113
datetime.timedelta(1, 200)
129
>>> # Can not be empty:
130
>>> rfc3339_duration_to_delta("")
131
Traceback (most recent call last):
132
...
133
ValueError: Invalid RFC 3339 duration: u''
134
>>> # Must start with "P":
135
>>> rfc3339_duration_to_delta("1D")
136
Traceback (most recent call last):
137
...
138
ValueError: Invalid RFC 3339 duration: u'1D'
139
>>> # Must use correct order
140
>>> rfc3339_duration_to_delta("PT1S2M")
141
Traceback (most recent call last):
142
...
143
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
144
>>> # Time needs time marker
145
>>> rfc3339_duration_to_delta("P1H2S")
146
Traceback (most recent call last):
147
...
148
ValueError: Invalid RFC 3339 duration: u'P1H2S'
149
>>> # Weeks can not be combined with anything else
150
>>> rfc3339_duration_to_delta("P1D2W")
151
Traceback (most recent call last):
152
...
153
ValueError: Invalid RFC 3339 duration: u'P1D2W'
154
>>> rfc3339_duration_to_delta("P2W2H")
155
Traceback (most recent call last):
156
...
157
ValueError: Invalid RFC 3339 duration: u'P2W2H'
158
114
"""
159
115
160
116
# Parsing an RFC 3339 duration with regular expressions is not
239
195
240
196
def string_to_delta(interval):
241
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)
242
211
"""
243
212
244
213
try:
245
214
return rfc3339_duration_to_delta(interval)
246
except ValueError as e:
247
log.warning("%s - Parsing as pre-1.6.1 interval instead",
248
' '.join(e.args))
249
return parse_pre_1_6_1_interval(interval)
250
251
252
def parse_pre_1_6_1_interval(interval):
253
"""Parse an interval string as documented by Mandos before 1.6.1, and
254
return a datetime.timedelta
255
>>> parse_pre_1_6_1_interval('7d')
256
datetime.timedelta(7)
257
>>> parse_pre_1_6_1_interval('60s')
258
datetime.timedelta(0, 60)
259
>>> parse_pre_1_6_1_interval('60m')
260
datetime.timedelta(0, 3600)
261
>>> parse_pre_1_6_1_interval('24h')
262
datetime.timedelta(1)
263
>>> parse_pre_1_6_1_interval('1w')
264
datetime.timedelta(7)
265
>>> parse_pre_1_6_1_interval('5m 30s')
266
datetime.timedelta(0, 330)
267
>>> parse_pre_1_6_1_interval('')
268
datetime.timedelta(0)
269
>>> # Ignore unknown characters, allow any order and repetitions
270
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
271
datetime.timedelta(2, 480, 18000)
272
273
"""
215
except ValueError:
216
pass
274
217
275
218
value = datetime.timedelta(0)
276
219
regexp = re.compile(r"(\d+)([dsmhw]?)")
292
235
293
236
294
237
def print_clients(clients, keywords):
295
print('\n'.join(table_rows_of_clients(clients, keywords)))
296
297
def table_rows_of_clients(clients, keywords):
298
238
def valuetostring(value, keyword):
299
if isinstance(value, dbus.Boolean):
239
if type(value) is dbus.Boolean:
300
240
return "Yes" if value else "No"
301
241
if keyword in ("Timeout", "Interval", "ApprovalDelay",
302
242
"ApprovalDuration", "ExtendedTimeout"):
310
250
for client in clients)),
311
251
key=key)
312
252
for key in keywords)
313
# Start with header line
314
rows = [format_string.format(**tablewords)]
253
# Print header line
254
print(format_string.format(**tablewords))
315
255
for client in clients:
316
rows.append(format_string
317
.format(**{key: valuetostring(client[key], key)
318
for key in keywords}))
319
return rows
256
print(format_string
257
.format(**{key: valuetostring(client[key], key)
258
for key in keywords}))
320
259
321
260
322
261
def has_actions(options):
406
345
if options.all and not has_actions(options):
407
346
parser.error("--all requires an action.")
408
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
409
353
try:
410
354
bus = dbus.SystemBus()
411
355
mandos_dbus_objc = bus.get_object(busname, server_path)
412
356
except dbus.exceptions.DBusException:
413
log.critical("Could not connect to Mandos server")
357
print("Could not connect to Mandos server", file=sys.stderr)
414
358
sys.exit(1)
415
359
416
360
mandos_serv = dbus.Interface(mandos_dbus_objc,
435
379
os.dup2(stderrcopy, sys.stderr.fileno())
436
380
os.close(stderrcopy)
437
381
except dbus.exceptions.DBusException as e:
438
log.critical("Failed to access Mandos server through D-Bus:"
439
"\n%s", e)
382
print("Access denied: "
383
"Accessing mandos server through D-Bus: {}".format(e),
384
file=sys.stderr)
440
385
sys.exit(1)
441
386
442
387
# Compile dict of (clients: properties) to process
453
398
clients[client_objc] = client
454
399
break
455
400
else:
456
log.critical("Client not found on server: %r", name)
401
print("Client not found on server: {!r}"
402
.format(name), file=sys.stderr)
457
403
sys.exit(1)
458
404
459
405
if not has_actions(options) and clients:
547
493
client.Approve(dbus.Boolean(False),
548
494
dbus_interface=client_interface)
549
495
550
551
class Test_milliseconds_to_string(unittest.TestCase):
552
def test_all(self):
553
self.assertEqual(milliseconds_to_string(93785000),
554
"1T02:03:05")
555
def test_no_days(self):
556
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
557
def test_all_zero(self):
558
self.assertEqual(milliseconds_to_string(0), "00:00:00")
559
def test_no_fractional_seconds(self):
560
self.assertEqual(milliseconds_to_string(400), "00:00:00")
561
self.assertEqual(milliseconds_to_string(900), "00:00:00")
562
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
563
564
class Test_string_to_delta(unittest.TestCase):
565
def test_handles_basic_rfc3339(self):
566
self.assertEqual(string_to_delta("PT2H"),
567
datetime.timedelta(0, 7200))
568
def test_falls_back_to_pre_1_6_1_with_warning(self):
569
# assertLogs only exists in Python 3.4
570
if hasattr(self, "assertLogs"):
571
with self.assertLogs(log, logging.WARNING):
572
value = string_to_delta("2h")
573
else:
574
value = string_to_delta("2h")
575
self.assertEqual(value, datetime.timedelta(0, 7200))
576
577
class Test_table_rows_of_clients(unittest.TestCase):
578
def setUp(self):
579
global tablewords
580
self.old_tablewords = tablewords
581
tablewords = {
582
"Attr1": "X",
583
"AttrTwo": "Yy",
584
"AttrThree": "Zzz",
585
"Bool": "A D-BUS Boolean",
586
"NonDbusBoolean": "A Non-D-BUS Boolean",
587
"Integer": "An Integer",
588
"Timeout": "Timedelta 1",
589
"Interval": "Timedelta 2",
590
"ApprovalDelay": "Timedelta 3",
591
"ApprovalDuration": "Timedelta 4",
592
"ExtendedTimeout": "Timedelta 5",
593
"String": "A String",
594
}
595
self.keywords = ["Attr1", "AttrTwo"]
596
self.clients = [
597
{
598
"Attr1": "x1",
599
"AttrTwo": "y1",
600
"AttrThree": "z1",
601
"Bool": dbus.Boolean(False),
602
"NonDbusBoolean": False,
603
"Integer": 0,
604
"Timeout": 0,
605
"Interval": 1000,
606
"ApprovalDelay": 2000,
607
"ApprovalDuration": 3000,
608
"ExtendedTimeout": 4000,
609
"String": "",
610
},
611
{
612
"Attr1": "x2",
613
"AttrTwo": "y2",
614
"AttrThree": "z2",
615
"Bool": dbus.Boolean(True),
616
"NonDbusBoolean": True,
617
"Integer": 1,
618
"Timeout": 93785000,
619
"Interval": 93786000,
620
"ApprovalDelay": 93787000,
621
"ApprovalDuration": 93788000,
622
"ExtendedTimeout": 93789000,
623
"String": "A huge string which will not fit," * 10,
624
},
625
]
626
def tearDown(self):
627
global tablewords
628
tablewords = self.old_tablewords
629
def test_short_header(self):
630
rows = table_rows_of_clients(self.clients, self.keywords)
631
expected_rows = [
632
"X Yy",
633
"x1 y1",
634
"x2 y2"]
635
self.assertEqual(rows, expected_rows)
636
def test_booleans(self):
637
keywords = ["Bool", "NonDbusBoolean"]
638
rows = table_rows_of_clients(self.clients, keywords)
639
expected_rows = [
640
"A D-BUS Boolean A Non-D-BUS Boolean",
641
"No False ",
642
"Yes True ",
643
]
644
self.assertEqual(rows, expected_rows)
645
def test_milliseconds_detection(self):
646
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
647
"ApprovalDuration", "ExtendedTimeout"]
648
rows = table_rows_of_clients(self.clients, keywords)
649
expected_rows = ("""
650
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
651
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
652
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
653
"""
654
).splitlines()[1:]
655
self.assertEqual(rows, expected_rows)
656
def test_empty_and_long_string_values(self):
657
keywords = ["String"]
658
rows = table_rows_of_clients(self.clients, keywords)
659
expected_rows = ("""
660
A String
661
662
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,
663
"""
664
).splitlines()[1:]
665
self.assertEqual(rows, expected_rows)
666
667
668
669
def should_only_run_tests():
670
parser = argparse.ArgumentParser(add_help=False)
671
parser.add_argument("--check", action='store_true')
672
args, unknown_args = parser.parse_known_args()
673
run_tests = args.check
674
if run_tests:
675
# Remove --check argument from sys.argv
676
sys.argv[1:] = unknown_args
677
return run_tests
678
679
# Add all tests from doctest strings
680
def load_tests(loader, tests, none):
681
import doctest
682
tests.addTests(doctest.DocTestSuite())
683
return tests
684
496
685
497
if __name__ == "__main__":
686
if should_only_run_tests():
687
# Call using ./tdd-python-script --check [--verbose]
688
unittest.main()
689
else:
690
main()
498
main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0