/mandos/release

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

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-02-13 08:55:24 UTC
  • Revision ID: teddy@recompile.se-20190213085524-qlzqay28sjfspl4z
Tags: version-1.8.3-3
http://bugs.debian.org/922202
* debian/changelog (1.8.3-3): Add new entry.

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; 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 -*-
3
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
40
40
import os
41
41
import collections
42
42
import json
43
 
import unittest
44
 
import logging
45
43
 
46
44
import dbus
47
45
 
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
 
 
59
46
if sys.version_info.major == 2:
60
47
    str = unicode
61
48
 
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):
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'
158
114
    """
159
115
 
160
116
    # Parsing an RFC 3339 duration with regular expressions is not
239
195
 
240
196
def string_to_delta(interval):
241
197
    """Parse a string and return a datetime.timedelta
 
198
 
 
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)
242
211
    """
243
212
 
244
213
    try:
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",
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
 
    """
 
215
    except ValueError:
 
216
        pass
274
217
 
275
218
    value = datetime.timedelta(0)
276
219
    regexp = re.compile(r"(\d+)([dsmhw]?)")
293
236
 
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.")
404
347
 
 
348
    if options.check:
 
349
        import doctest
 
350
        fail_count, test_count = doctest.testmod()
 
351
        sys.exit(os.EX_OK if fail_count == 0 else 1)
 
352
 
405
353
    try:
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)
410
358
        sys.exit(1)
411
359
 
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:"
435
 
                     "\n%s", e)
 
382
        print("Access denied: "
 
383
              "Accessing mandos server through D-Bus: {}".format(e),
 
384
              file=sys.stderr)
436
385
        sys.exit(1)
437
386
 
438
387
    # Compile dict of (clients: properties) to process
449
398
                    clients[client_objc] = client
450
399
                    break
451
400
            else:
452
 
                log.critical("Client not found on server: %r", name)
 
401
                print("Client not found on server: {!r}"
 
402
                      .format(name), file=sys.stderr)
453
403
                sys.exit(1)
454
404
 
455
405
    if not has_actions(options) and clients:
543
493
                client.Approve(dbus.Boolean(False),
544
494
                               dbus_interface=client_interface)
545
495
 
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
589
496
 
590
497
if __name__ == "__main__":
591
 
    if should_only_run_tests():
592
 
        # Call using ./tdd-python-script --check [--verbose]
593
 
        unittest.main()
594
 
    else:
595
 
        main()
 
498
    main()