2
# -*- mode: python; coding: utf-8 -*-
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*")))); -*-
4
4
# Mandos Monitor - Control and monitor the Mandos server
6
# Copyright © 2008-2017 Teddy Hogeborn
7
# Copyright © 2008-2017 Björn Påhlsson
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
6
# Copyright © 2008-2019 Teddy Hogeborn
7
# Copyright © 2008-2019 Björn Påhlsson
9
# This file is part of Mandos.
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.
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.
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
# along with Mandos. If not, see <http://www.gnu.org/licenses/>.
23
24
# Contact the authors at <mandos@recompile.se>.
48
# Show warnings by default
49
if not sys.warnoptions:
51
warnings.simplefilter("default")
53
log = logging.getLogger(sys.argv[0])
54
logging.basicConfig(level="INFO", # Show info level messages
55
format="%(message)s") # Show basic log messages
57
logging.captureWarnings(True) # Show warnings via the logging system
45
59
if sys.version_info.major == 2:
48
62
locale.setlocale(locale.LC_ALL, "")
54
"LastCheckedOK": "Last Successful Check",
55
"LastApprovalRequest": "Last Approval Request",
57
"Interval": "Interval",
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",
67
"ExtendedTimeout": "Extended Timeout",
69
"LastCheckerStatus": "Last Checker Status",
71
64
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
72
65
domain = "se.recompile"
73
66
busname = domain + ".Mandos"
75
68
server_interface = domain + ".Mandos"
76
69
client_interface = domain + ".Mandos.Client"
109
104
datetime.timedelta(0, 330)
110
105
>>> rfc3339_duration_to_delta("P1DT3M20S")
111
106
datetime.timedelta(1, 200)
107
>>> # Can not be empty:
108
>>> rfc3339_duration_to_delta("")
109
Traceback (most recent call last):
111
ValueError: Invalid RFC 3339 duration: u''
112
>>> # Must start with "P":
113
>>> rfc3339_duration_to_delta("1D")
114
Traceback (most recent call last):
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):
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):
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):
131
ValueError: Invalid RFC 3339 duration: u'P1D2W'
132
>>> rfc3339_duration_to_delta("P2W2H")
133
Traceback (most recent call last):
135
ValueError: Invalid RFC 3339 duration: u'P2W2H'
114
138
# Parsing an RFC 3339 duration with regular expressions is not
194
218
def string_to_delta(interval):
195
219
"""Parse a string and return a datetime.timedelta
197
>>> string_to_delta('7d')
198
datetime.timedelta(7)
199
>>> string_to_delta('60s')
200
datetime.timedelta(0, 60)
201
>>> string_to_delta('60m')
202
datetime.timedelta(0, 3600)
203
>>> string_to_delta('24h')
204
datetime.timedelta(1)
205
>>> string_to_delta('1w')
206
datetime.timedelta(7)
207
>>> string_to_delta('5m 30s')
208
datetime.timedelta(0, 330)
212
223
return rfc3339_duration_to_delta(interval)
224
except ValueError as e:
225
log.warning("%s - Parsing as pre-1.6.1 interval instead",
227
return parse_pre_1_6_1_interval(interval)
230
def parse_pre_1_6_1_interval(interval):
231
"""Parse an interval string as documented by Mandos before 1.6.1, and
232
return a datetime.timedelta
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)
216
253
value = datetime.timedelta(0)
217
254
regexp = re.compile(r"(\d+)([dsmhw]?)")
235
272
def print_clients(clients, keywords):
273
print('\n'.join(TableOfClients(clients, keywords).rows()))
275
class TableOfClients(object):
278
"Enabled": "Enabled",
279
"Timeout": "Timeout",
280
"LastCheckedOK": "Last Successful Check",
281
"LastApprovalRequest": "Last Approval Request",
282
"Created": "Created",
283
"Interval": "Interval",
285
"Fingerprint": "Fingerprint",
287
"CheckerRunning": "Check Is Running",
288
"LastEnabled": "Last Enabled",
289
"ApprovalPending": "Approval Is Pending",
290
"ApprovedByDefault": "Approved By Default",
291
"ApprovalDelay": "Approval Delay",
292
"ApprovalDuration": "Approval Duration",
293
"Checker": "Checker",
294
"ExtendedTimeout": "Extended Timeout",
295
"Expires": "Expires",
296
"LastCheckerStatus": "Last Checker Status",
299
def __init__(self, clients, keywords, tablewords=None):
300
self.clients = clients
301
self.keywords = keywords
302
if tablewords is not None:
303
self.tablewords = tablewords
236
306
def valuetostring(value, keyword):
237
if type(value) is dbus.Boolean:
307
if isinstance(value, dbus.Boolean):
238
308
return "Yes" if value else "No"
239
309
if keyword in ("Timeout", "Interval", "ApprovalDelay",
240
310
"ApprovalDuration", "ExtendedTimeout"):
241
311
return milliseconds_to_string(value)
242
312
return str(value)
244
# Create format string to print table rows
245
format_string = " ".join("{{{key}:{width}}}".format(
246
width=max(len(tablewords[key]),
247
max(len(valuetostring(client[key], key))
248
for client in clients)),
252
print(format_string.format(**tablewords))
253
for client in clients:
255
.format(**{key: valuetostring(client[key], key)
256
for key in keywords}))
315
# Create format string to format table rows
316
format_string = " ".join("{{{key}:{width}}}".format(
317
width=max(len(self.tablewords[key]),
318
max(len(self.valuetostring(client[key], key))
319
for client in self.clients)),
321
for key in self.keywords)
322
# Start with header line
323
rows = [format_string.format(**self.tablewords)]
324
for client in self.clients:
325
rows.append(format_string
326
.format(**{key: self.valuetostring(client[key], key)
327
for key in self.keywords}))
259
331
def has_actions(options):
396
462
clients[client_objc] = client
399
print("Client not found on server: {!r}"
400
.format(name), file=sys.stderr)
465
log.critical("Client not found on server: %r", name)
403
468
if not has_actions(options) and clients:
404
469
if options.verbose or options.dump_json:
405
470
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
406
"Created", "Interval", "Host", "Fingerprint",
407
"CheckerRunning", "LastEnabled",
408
"ApprovalPending", "ApprovedByDefault",
409
"LastApprovalRequest", "ApprovalDelay",
410
"ApprovalDuration", "Checker",
411
"ExtendedTimeout", "Expires",
471
"Created", "Interval", "Host", "KeyID",
472
"Fingerprint", "CheckerRunning",
473
"LastEnabled", "ApprovalPending",
474
"ApprovedByDefault", "LastApprovalRequest",
475
"ApprovalDelay", "ApprovalDuration",
476
"Checker", "ExtendedTimeout", "Expires",
412
477
"LastCheckerStatus")
414
479
keywords = defaultkeywords
491
556
client.Approve(dbus.Boolean(False),
492
557
dbus_interface=client_interface)
560
class Test_milliseconds_to_string(unittest.TestCase):
562
self.assertEqual(milliseconds_to_string(93785000),
564
def test_no_days(self):
565
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
566
def test_all_zero(self):
567
self.assertEqual(milliseconds_to_string(0), "00:00:00")
568
def test_no_fractional_seconds(self):
569
self.assertEqual(milliseconds_to_string(400), "00:00:00")
570
self.assertEqual(milliseconds_to_string(900), "00:00:00")
571
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
573
class Test_string_to_delta(unittest.TestCase):
574
def test_handles_basic_rfc3339(self):
575
self.assertEqual(string_to_delta("PT2H"),
576
datetime.timedelta(0, 7200))
577
def test_falls_back_to_pre_1_6_1_with_warning(self):
578
# assertLogs only exists in Python 3.4
579
if hasattr(self, "assertLogs"):
580
with self.assertLogs(log, logging.WARNING):
581
value = string_to_delta("2h")
583
value = string_to_delta("2h")
584
self.assertEqual(value, datetime.timedelta(0, 7200))
586
class Test_TableOfClients(unittest.TestCase):
592
"Bool": "A D-BUS Boolean",
593
"NonDbusBoolean": "A Non-D-BUS Boolean",
594
"Integer": "An Integer",
595
"Timeout": "Timedelta 1",
596
"Interval": "Timedelta 2",
597
"ApprovalDelay": "Timedelta 3",
598
"ApprovalDuration": "Timedelta 4",
599
"ExtendedTimeout": "Timedelta 5",
600
"String": "A String",
602
self.keywords = ["Attr1", "AttrTwo"]
608
"Bool": dbus.Boolean(False),
609
"NonDbusBoolean": False,
613
"ApprovalDelay": 2000,
614
"ApprovalDuration": 3000,
615
"ExtendedTimeout": 4000,
622
"Bool": dbus.Boolean(True),
623
"NonDbusBoolean": True,
626
"Interval": 93786000,
627
"ApprovalDelay": 93787000,
628
"ApprovalDuration": 93788000,
629
"ExtendedTimeout": 93789000,
630
"String": "A huge string which will not fit," * 10,
633
def test_short_header(self):
634
rows = TableOfClients(self.clients, self.keywords,
635
self.tablewords).rows()
640
self.assertEqual(rows, expected_rows)
641
def test_booleans(self):
642
keywords = ["Bool", "NonDbusBoolean"]
643
rows = TableOfClients(self.clients, keywords,
644
self.tablewords).rows()
646
"A D-BUS Boolean A Non-D-BUS Boolean",
650
self.assertEqual(rows, expected_rows)
651
def test_milliseconds_detection(self):
652
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
653
"ApprovalDuration", "ExtendedTimeout"]
654
rows = TableOfClients(self.clients, keywords,
655
self.tablewords).rows()
657
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
658
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
659
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
662
self.assertEqual(rows, expected_rows)
663
def test_empty_and_long_string_values(self):
664
keywords = ["String"]
665
rows = TableOfClients(self.clients, keywords,
666
self.tablewords).rows()
670
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,
673
self.assertEqual(rows, expected_rows)
677
def should_only_run_tests():
678
parser = argparse.ArgumentParser(add_help=False)
679
parser.add_argument("--check", action='store_true')
680
args, unknown_args = parser.parse_known_args()
681
run_tests = args.check
683
# Remove --check argument from sys.argv
684
sys.argv[1:] = unknown_args
687
# Add all tests from doctest strings
688
def load_tests(loader, tests, none):
690
tests.addTests(doctest.DocTestSuite())
495
693
if __name__ == "__main__":
694
if should_only_run_tests():
695
# Call using ./tdd-python-script --check [--verbose]