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
6
# Copyright © 2008-2019 Teddy Hogeborn
7
# Copyright © 2008-2019 Björn Påhlsson
9
9
# This file is part of Mandos.
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
46
59
if sys.version_info.major == 2:
58
71
"Interval": "Interval",
60
73
"Fingerprint": "Fingerprint",
61
75
"CheckerRunning": "Check Is Running",
62
76
"LastEnabled": "Last Enabled",
63
77
"ApprovalPending": "Approval Is Pending",
102
116
datetime.timedelta(0, 60)
103
117
>>> rfc3339_duration_to_delta("PT60M")
104
118
datetime.timedelta(0, 3600)
119
>>> rfc3339_duration_to_delta("P60M")
120
datetime.timedelta(1680)
105
121
>>> rfc3339_duration_to_delta("PT24H")
106
122
datetime.timedelta(1)
107
123
>>> rfc3339_duration_to_delta("P1W")
110
126
datetime.timedelta(0, 330)
111
127
>>> rfc3339_duration_to_delta("P1DT3M20S")
112
128
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'
115
160
# Parsing an RFC 3339 duration with regular expressions is not
195
240
def string_to_delta(interval):
196
241
"""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)
213
245
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)
217
275
value = datetime.timedelta(0)
218
276
regexp = re.compile(r"(\d+)([dsmhw]?)")
236
294
def print_clients(clients, keywords):
237
295
def valuetostring(value, keyword):
238
if type(value) is dbus.Boolean:
296
if isinstance(value, dbus.Boolean):
239
297
return "Yes" if value else "No"
240
298
if keyword in ("Timeout", "Interval", "ApprovalDelay",
241
299
"ApprovalDuration", "ExtendedTimeout"):
344
402
if options.all and not has_actions(options):
345
403
parser.error("--all requires an action.")
349
fail_count, test_count = doctest.testmod()
350
sys.exit(os.EX_OK if fail_count == 0 else 1)
353
406
bus = dbus.SystemBus()
354
407
mandos_dbus_objc = bus.get_object(busname, server_path)
355
408
except dbus.exceptions.DBusException:
356
print("Could not connect to Mandos server", file=sys.stderr)
409
log.critical("Could not connect to Mandos server")
359
412
mandos_serv = dbus.Interface(mandos_dbus_objc,
378
431
os.dup2(stderrcopy, sys.stderr.fileno())
379
432
os.close(stderrcopy)
380
433
except dbus.exceptions.DBusException as e:
381
print("Access denied: "
382
"Accessing mandos server through D-Bus: {}".format(e),
434
log.critical("Failed to access Mandos server through D-Bus:"
386
438
# Compile dict of (clients: properties) to process
397
449
clients[client_objc] = client
400
print("Client not found on server: {!r}"
401
.format(name), file=sys.stderr)
452
log.critical("Client not found on server: %r", name)
404
455
if not has_actions(options) and clients:
405
456
if options.verbose or options.dump_json:
406
457
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",
458
"Created", "Interval", "Host", "KeyID",
459
"Fingerprint", "CheckerRunning",
460
"LastEnabled", "ApprovalPending",
461
"ApprovedByDefault", "LastApprovalRequest",
462
"ApprovalDelay", "ApprovalDuration",
463
"Checker", "ExtendedTimeout", "Expires",
413
464
"LastCheckerStatus")
415
466
keywords = defaultkeywords
492
543
client.Approve(dbus.Boolean(False),
493
544
dbus_interface=client_interface)
547
class Test_milliseconds_to_string(unittest.TestCase):
549
self.assertEqual(milliseconds_to_string(93785000),
551
def test_no_days(self):
552
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
553
def test_all_zero(self):
554
self.assertEqual(milliseconds_to_string(0), "00:00:00")
555
def test_no_fractional_seconds(self):
556
self.assertEqual(milliseconds_to_string(400), "00:00:00")
557
self.assertEqual(milliseconds_to_string(900), "00:00:00")
558
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
560
class Test_string_to_delta(unittest.TestCase):
561
def test_handles_basic_rfc3339(self):
562
self.assertEqual(string_to_delta("PT2H"),
563
datetime.timedelta(0, 7200))
564
def test_falls_back_to_pre_1_6_1_with_warning(self):
565
# assertLogs only exists in Python 3.4
566
if hasattr(self, "assertLogs"):
567
with self.assertLogs(log, logging.WARNING):
568
value = string_to_delta("2h")
570
value = string_to_delta("2h")
571
self.assertEqual(value, datetime.timedelta(0, 7200))
574
def should_only_run_tests():
575
parser = argparse.ArgumentParser(add_help=False)
576
parser.add_argument("--check", action='store_true')
577
args, unknown_args = parser.parse_known_args()
578
run_tests = args.check
580
# Remove --check argument from sys.argv
581
sys.argv[1:] = unknown_args
584
# Add all tests from doctest strings
585
def load_tests(loader, tests, none):
587
tests.addTests(doctest.DocTestSuite())
496
590
if __name__ == "__main__":
591
if should_only_run_tests():
592
# Call using ./tdd-python-script --check [--verbose]