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
6
# Copyright © 2008-2018 Teddy Hogeborn
7
# Copyright © 2008-2018 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
59
46
if sys.version_info.major == 2:
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):
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
116
# Parsing an RFC 3339 duration with regular expressions is not
240
196
def string_to_delta(interval):
241
197
"""Parse a string and return a datetime.timedelta
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)
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",
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
218
value = datetime.timedelta(0)
276
219
regexp = re.compile(r"(\d+)([dsmhw]?)")
294
237
def print_clients(clients, keywords):
295
238
def valuetostring(value, keyword):
296
if isinstance(value, dbus.Boolean):
239
if type(value) is dbus.Boolean:
297
240
return "Yes" if value else "No"
298
241
if keyword in ("Timeout", "Interval", "ApprovalDelay",
299
242
"ApprovalDuration", "ExtendedTimeout"):
402
345
if options.all and not has_actions(options):
403
346
parser.error("--all requires an action.")
350
fail_count, test_count = doctest.testmod()
351
sys.exit(os.EX_OK if fail_count == 0 else 1)
406
354
bus = dbus.SystemBus()
407
355
mandos_dbus_objc = bus.get_object(busname, server_path)
408
356
except dbus.exceptions.DBusException:
409
log.critical("Could not connect to Mandos server")
357
print("Could not connect to Mandos server", file=sys.stderr)
412
360
mandos_serv = dbus.Interface(mandos_dbus_objc,
431
379
os.dup2(stderrcopy, sys.stderr.fileno())
432
380
os.close(stderrcopy)
433
381
except dbus.exceptions.DBusException as e:
434
log.critical("Failed to access Mandos server through D-Bus:"
382
print("Access denied: "
383
"Accessing mandos server through D-Bus: {}".format(e),
438
387
# Compile dict of (clients: properties) to process
449
398
clients[client_objc] = client
452
log.critical("Client not found on server: %r", name)
401
print("Client not found on server: {!r}"
402
.format(name), file=sys.stderr)
455
405
if not has_actions(options) and clients:
543
493
client.Approve(dbus.Boolean(False),
544
494
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")
561
def should_only_run_tests():
562
parser = argparse.ArgumentParser(add_help=False)
563
parser.add_argument("--check", action='store_true')
564
args, unknown_args = parser.parse_known_args()
565
run_tests = args.check
567
# Remove --check argument from sys.argv
568
sys.argv[1:] = unknown_args
571
# Add all tests from doctest strings
572
def load_tests(loader, tests, none):
574
tests.addTests(doctest.DocTestSuite())
577
497
if __name__ == "__main__":
578
if should_only_run_tests():
579
# Call using ./tdd-python-script --check [--verbose]
added
removed
mandos-ctl
