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 -*-
4
4
# Mandos Monitor - Control and monitor the Mandos server
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
6
# Copyright © 2008-2016 Teddy Hogeborn
7
# Copyright © 2008-2016 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
13
11
# the Free Software Foundation, either version 3 of the License, or
14
12
# (at your option) any later version.
16
# Mandos is distributed in the hope that it will be useful, but
17
# WITHOUT ANY WARRANTY; without even the implied warranty of
14
# This program is distributed in the hope that it will be useful,
15
# but WITHOUT ANY WARRANTY; without even the implied warranty of
18
16
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19
17
# GNU General Public License for more details.
21
19
# You should have received a copy of the GNU General Public License
22
# along with Mandos. If not, see <http://www.gnu.org/licenses/>.
20
# along with this program. If not, see
21
# <http://www.gnu.org/licenses/>.
24
23
# Contact the authors at <mandos@recompile.se>.
126
110
datetime.timedelta(0, 330)
127
111
>>> rfc3339_duration_to_delta("P1DT3M20S")
128
112
datetime.timedelta(1, 200)
129
>>> # Can not be empty:
130
>>> rfc3339_duration_to_delta("")
131
Traceback (most recent call last):
133
ValueError: Invalid RFC 3339 duration: u''
134
>>> # Must start with "P":
135
>>> rfc3339_duration_to_delta("1D")
136
Traceback (most recent call last):
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):
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):
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):
153
ValueError: Invalid RFC 3339 duration: u'P1D2W'
154
>>> rfc3339_duration_to_delta("P2W2H")
155
Traceback (most recent call last):
157
ValueError: Invalid RFC 3339 duration: u'P2W2H'
160
115
# Parsing an RFC 3339 duration with regular expressions is not
240
195
def string_to_delta(interval):
241
196
"""Parse a string and return a datetime.timedelta
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)
245
213
return rfc3339_duration_to_delta(interval)
246
except ValueError as e:
247
log.warning("%s - Parsing as pre-1.6.1 interval instead",
249
return parse_pre_1_6_1_interval(interval)
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)
275
217
value = datetime.timedelta(0)
276
218
regexp = re.compile(r"(\d+)([dsmhw]?)")
294
236
def print_clients(clients, keywords):
295
print('\n'.join(TableOfClients(clients, keywords).rows()))
297
class TableOfClients(object):
298
def __init__(self, clients, keywords):
299
self.clients = clients
300
self.keywords = keywords
303
237
def valuetostring(value, keyword):
304
if isinstance(value, dbus.Boolean):
238
if type(value) is dbus.Boolean:
305
239
return "Yes" if value else "No"
306
240
if keyword in ("Timeout", "Interval", "ApprovalDelay",
307
241
"ApprovalDuration", "ExtendedTimeout"):
308
242
return milliseconds_to_string(value)
309
243
return str(value)
312
# Create format string to format table rows
313
format_string = " ".join("{{{key}:{width}}}".format(
314
width=max(len(tablewords[key]),
315
max(len(self.valuetostring(client[key], key))
316
for client in self.clients)),
318
for key in self.keywords)
319
# Start with header line
320
rows = [format_string.format(**tablewords)]
321
for client in self.clients:
322
rows.append(format_string
323
.format(**{key: self.valuetostring(client[key], key)
324
for key in self.keywords}))
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)),
253
print(format_string.format(**tablewords))
254
for client in clients:
256
.format(**{key: valuetostring(client[key], key)
257
for key in keywords}))
328
260
def has_actions(options):
459
396
clients[client_objc] = client
462
log.critical("Client not found on server: %r", name)
399
print("Client not found on server: {!r}"
400
.format(name), file=sys.stderr)
465
403
if not has_actions(options) and clients:
466
404
if options.verbose or options.dump_json:
467
405
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
468
"Created", "Interval", "Host", "KeyID",
469
"Fingerprint", "CheckerRunning",
470
"LastEnabled", "ApprovalPending",
471
"ApprovedByDefault", "LastApprovalRequest",
472
"ApprovalDelay", "ApprovalDuration",
473
"Checker", "ExtendedTimeout", "Expires",
406
"Created", "Interval", "Host", "Fingerprint",
407
"CheckerRunning", "LastEnabled",
408
"ApprovalPending", "ApprovedByDefault",
409
"LastApprovalRequest", "ApprovalDelay",
410
"ApprovalDuration", "Checker",
411
"ExtendedTimeout", "Expires",
474
412
"LastCheckerStatus")
476
414
keywords = defaultkeywords
553
491
client.Approve(dbus.Boolean(False),
554
492
dbus_interface=client_interface)
557
class Test_milliseconds_to_string(unittest.TestCase):
559
self.assertEqual(milliseconds_to_string(93785000),
561
def test_no_days(self):
562
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
563
def test_all_zero(self):
564
self.assertEqual(milliseconds_to_string(0), "00:00:00")
565
def test_no_fractional_seconds(self):
566
self.assertEqual(milliseconds_to_string(400), "00:00:00")
567
self.assertEqual(milliseconds_to_string(900), "00:00:00")
568
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
570
class Test_string_to_delta(unittest.TestCase):
571
def test_handles_basic_rfc3339(self):
572
self.assertEqual(string_to_delta("PT2H"),
573
datetime.timedelta(0, 7200))
574
def test_falls_back_to_pre_1_6_1_with_warning(self):
575
# assertLogs only exists in Python 3.4
576
if hasattr(self, "assertLogs"):
577
with self.assertLogs(log, logging.WARNING):
578
value = string_to_delta("2h")
580
value = string_to_delta("2h")
581
self.assertEqual(value, datetime.timedelta(0, 7200))
583
class Test_TableOfClients(unittest.TestCase):
586
self.old_tablewords = tablewords
591
"Bool": "A D-BUS Boolean",
592
"NonDbusBoolean": "A Non-D-BUS Boolean",
593
"Integer": "An Integer",
594
"Timeout": "Timedelta 1",
595
"Interval": "Timedelta 2",
596
"ApprovalDelay": "Timedelta 3",
597
"ApprovalDuration": "Timedelta 4",
598
"ExtendedTimeout": "Timedelta 5",
599
"String": "A String",
601
self.keywords = ["Attr1", "AttrTwo"]
607
"Bool": dbus.Boolean(False),
608
"NonDbusBoolean": False,
612
"ApprovalDelay": 2000,
613
"ApprovalDuration": 3000,
614
"ExtendedTimeout": 4000,
621
"Bool": dbus.Boolean(True),
622
"NonDbusBoolean": True,
625
"Interval": 93786000,
626
"ApprovalDelay": 93787000,
627
"ApprovalDuration": 93788000,
628
"ExtendedTimeout": 93789000,
629
"String": "A huge string which will not fit," * 10,
634
tablewords = self.old_tablewords
635
def test_short_header(self):
636
rows = TableOfClients(self.clients, self.keywords).rows()
641
self.assertEqual(rows, expected_rows)
642
def test_booleans(self):
643
keywords = ["Bool", "NonDbusBoolean"]
644
rows = TableOfClients(self.clients, keywords).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).rows()
656
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
657
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
658
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
661
self.assertEqual(rows, expected_rows)
662
def test_empty_and_long_string_values(self):
663
keywords = ["String"]
664
rows = TableOfClients(self.clients, keywords).rows()
668
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,
671
self.assertEqual(rows, expected_rows)
675
def should_only_run_tests():
676
parser = argparse.ArgumentParser(add_help=False)
677
parser.add_argument("--check", action='store_true')
678
args, unknown_args = parser.parse_known_args()
679
run_tests = args.check
681
# Remove --check argument from sys.argv
682
sys.argv[1:] = unknown_args
685
# Add all tests from doctest strings
686
def load_tests(loader, tests, none):
688
tests.addTests(doctest.DocTestSuite())
691
495
if __name__ == "__main__":
692
if should_only_run_tests():
693
# Call using ./tdd-python-script --check [--verbose]