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:
57
71
"Interval": "Interval",
59
73
"Fingerprint": "Fingerprint",
60
75
"CheckerRunning": "Check Is Running",
61
76
"LastEnabled": "Last Enabled",
62
77
"ApprovalPending": "Approval Is Pending",
101
116
datetime.timedelta(0, 60)
102
117
>>> rfc3339_duration_to_delta("PT60M")
103
118
datetime.timedelta(0, 3600)
119
>>> rfc3339_duration_to_delta("P60M")
120
datetime.timedelta(1680)
104
121
>>> rfc3339_duration_to_delta("PT24H")
105
122
datetime.timedelta(1)
106
123
>>> rfc3339_duration_to_delta("P1W")
109
126
datetime.timedelta(0, 330)
110
127
>>> rfc3339_duration_to_delta("P1DT3M20S")
111
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'
114
160
# Parsing an RFC 3339 duration with regular expressions is not
194
240
def string_to_delta(interval):
195
241
"""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
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)
216
275
value = datetime.timedelta(0)
217
276
regexp = re.compile(r"(\d+)([dsmhw]?)")
235
294
def print_clients(clients, keywords):
236
295
def valuetostring(value, keyword):
237
if type(value) is dbus.Boolean:
296
if isinstance(value, dbus.Boolean):
238
297
return "Yes" if value else "No"
239
298
if keyword in ("Timeout", "Interval", "ApprovalDelay",
240
299
"ApprovalDuration", "ExtendedTimeout"):
343
402
if options.all and not has_actions(options):
344
403
parser.error("--all requires an action.")
348
fail_count, test_count = doctest.testmod()
349
sys.exit(os.EX_OK if fail_count == 0 else 1)
352
406
bus = dbus.SystemBus()
353
407
mandos_dbus_objc = bus.get_object(busname, server_path)
354
408
except dbus.exceptions.DBusException:
355
print("Could not connect to Mandos server", file=sys.stderr)
409
log.critical("Could not connect to Mandos server")
358
412
mandos_serv = dbus.Interface(mandos_dbus_objc,
377
431
os.dup2(stderrcopy, sys.stderr.fileno())
378
432
os.close(stderrcopy)
379
433
except dbus.exceptions.DBusException as e:
380
print("Access denied: "
381
"Accessing mandos server through D-Bus: {}".format(e),
434
log.critical("Failed to access Mandos server through D-Bus:"
385
438
# Compile dict of (clients: properties) to process
396
449
clients[client_objc] = client
399
print("Client not found on server: {!r}"
400
.format(name), file=sys.stderr)
452
log.critical("Client not found on server: %r", name)
403
455
if not has_actions(options) and clients:
404
456
if options.verbose or options.dump_json:
405
457
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",
458
"Created", "Interval", "Host", "KeyID",
459
"Fingerprint", "CheckerRunning",
460
"LastEnabled", "ApprovalPending",
461
"ApprovedByDefault", "LastApprovalRequest",
462
"ApprovalDelay", "ApprovalDuration",
463
"Checker", "ExtendedTimeout", "Expires",
412
464
"LastCheckerStatus")
414
466
keywords = defaultkeywords
491
543
client.Approve(dbus.Boolean(False),
492
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")
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())
495
577
if __name__ == "__main__":
578
if should_only_run_tests():
579
# Call using ./tdd-python-script --check [--verbose]