/mandos/trunk

To get this branch, use:
bzr branch http://bzr.recompile.se/loggerhead/mandos/trunk

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-02-28 22:17:56 UTC
  • Revision ID: teddy@recompile.se-20190228221756-qd6ynzhzatmp2g6a
mandos-ctl: Add more tests

* mandos-ctl (string_to_delta): Add some basic tests.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
1
#!/usr/bin/python
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*")))); -*-
3
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
6
 
# Copyright © 2008-2018 Teddy Hogeborn
7
 
# Copyright © 2008-2018 Björn Påhlsson
 
6
# Copyright © 2008-2019 Teddy Hogeborn
 
7
# Copyright © 2008-2019 Björn Påhlsson
8
8
#
9
9
# This file is part of Mandos.
10
10
#
40
40
import os
41
41
import collections
42
42
import json
 
43
import unittest
 
44
import logging
43
45
 
44
46
import dbus
45
47
 
 
48
# Show warnings by default
 
49
if not sys.warnoptions:
 
50
    import warnings
 
51
    warnings.simplefilter("default")
 
52
 
 
53
log = logging.getLogger(sys.argv[0])
 
54
logging.basicConfig(level="INFO", # Show info level messages
 
55
                    format="%(message)s") # Show basic log messages
 
56
 
 
57
logging.captureWarnings(True)   # Show warnings via the logging system
 
58
 
46
59
if sys.version_info.major == 2:
47
60
    str = unicode
48
61
 
58
71
    "Interval": "Interval",
59
72
    "Host": "Host",
60
73
    "Fingerprint": "Fingerprint",
 
74
    "KeyID": "Key ID",
61
75
    "CheckerRunning": "Check Is Running",
62
76
    "LastEnabled": "Last Enabled",
63
77
    "ApprovalPending": "Approval Is Pending",
75
89
server_path = "/"
76
90
server_interface = domain + ".Mandos"
77
91
client_interface = domain + ".Mandos.Client"
78
 
version = "1.7.19"
 
92
version = "1.8.3"
79
93
 
80
94
 
81
95
try:
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):
 
132
    ...
 
133
    ValueError: Invalid RFC 3339 duration: u''
 
134
    >>> # Must start with "P":
 
135
    >>> rfc3339_duration_to_delta("1D")
 
136
    Traceback (most recent call last):
 
137
    ...
 
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):
 
142
    ...
 
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):
 
147
    ...
 
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):
 
152
    ...
 
153
    ValueError: Invalid RFC 3339 duration: u'P1D2W'
 
154
    >>> rfc3339_duration_to_delta("P2W2H")
 
155
    Traceback (most recent call last):
 
156
    ...
 
157
    ValueError: Invalid RFC 3339 duration: u'P2W2H'
113
158
    """
114
159
 
115
160
    # Parsing an RFC 3339 duration with regular expressions is not
194
239
 
195
240
def string_to_delta(interval):
196
241
    """Parse a string and return a datetime.timedelta
197
 
 
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)
210
242
    """
211
243
 
212
244
    try:
213
245
        return rfc3339_duration_to_delta(interval)
214
 
    except ValueError:
215
 
        pass
 
246
    except ValueError as e:
 
247
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
 
248
                    ' '.join(e.args))
 
249
    return parse_pre_1_6_1_interval(interval)
 
250
 
 
251
 
 
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)
 
272
 
 
273
    """
216
274
 
217
275
    value = datetime.timedelta(0)
218
276
    regexp = re.compile(r"(\d+)([dsmhw]?)")
235
293
 
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.")
346
404
 
347
 
    if options.check:
348
 
        import doctest
349
 
        fail_count, test_count = doctest.testmod()
350
 
        sys.exit(os.EX_OK if fail_count == 0 else 1)
351
 
 
352
405
    try:
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")
357
410
        sys.exit(1)
358
411
 
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),
383
 
              file=sys.stderr)
 
434
        log.critical("Failed to access Mandos server through D-Bus:"
 
435
                     "\n%s", e)
384
436
        sys.exit(1)
385
437
 
386
438
    # Compile dict of (clients: properties) to process
397
449
                    clients[client_objc] = client
398
450
                    break
399
451
            else:
400
 
                print("Client not found on server: {!r}"
401
 
                      .format(name), file=sys.stderr)
 
452
                log.critical("Client not found on server: %r", name)
402
453
                sys.exit(1)
403
454
 
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")
414
465
        else:
415
466
            keywords = defaultkeywords
492
543
                client.Approve(dbus.Boolean(False),
493
544
                               dbus_interface=client_interface)
494
545
 
 
546
 
 
547
class Test_milliseconds_to_string(unittest.TestCase):
 
548
    def test_all(self):
 
549
        self.assertEqual(milliseconds_to_string(93785000),
 
550
                         "1T02:03:05")
 
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")
 
559
 
 
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")
 
569
        else:
 
570
            value = string_to_delta("2h")
 
571
        self.assertEqual(value, datetime.timedelta(0, 7200))
 
572
 
 
573
 
 
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
 
579
    if run_tests:
 
580
        # Remove --check argument from sys.argv
 
581
        sys.argv[1:] = unknown_args
 
582
    return run_tests
 
583
 
 
584
# Add all tests from doctest strings
 
585
def load_tests(loader, tests, none):
 
586
    import doctest
 
587
    tests.addTests(doctest.DocTestSuite())
 
588
    return tests
495
589
 
496
590
if __name__ == "__main__":
497
 
    main()
 
591
    if should_only_run_tests():
 
592
        # Call using ./tdd-python-script --check [--verbose]
 
593
        unittest.main()
 
594
    else:
 
595
        main()