/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-03-02 00:26:04 UTC
  • Revision ID: teddy@recompile.se-20190302002604-r9kpfc7n8qllwg62
mandos-ctl (print_clients): Factor out making strings from printing

* mandos-ctl (table_rows_of_clients): New.

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
#
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
 
103
116
    datetime.timedelta(0, 60)
104
117
    >>> rfc3339_duration_to_delta("PT60M")
105
118
    datetime.timedelta(0, 3600)
 
119
    >>> rfc3339_duration_to_delta("P60M")
 
120
    datetime.timedelta(1680)
106
121
    >>> rfc3339_duration_to_delta("PT24H")
107
122
    datetime.timedelta(1)
108
123
    >>> rfc3339_duration_to_delta("P1W")
111
126
    datetime.timedelta(0, 330)
112
127
    >>> rfc3339_duration_to_delta("P1DT3M20S")
113
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'
114
158
    """
115
159
 
116
160
    # Parsing an RFC 3339 duration with regular expressions is not
195
239
 
196
240
def string_to_delta(interval):
197
241
    """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)
211
242
    """
212
243
 
213
244
    try:
214
245
        return rfc3339_duration_to_delta(interval)
215
 
    except ValueError:
216
 
        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
    """
217
274
 
218
275
    value = datetime.timedelta(0)
219
276
    regexp = re.compile(r"(\d+)([dsmhw]?)")
235
292
 
236
293
 
237
294
def print_clients(clients, keywords):
 
295
    print('\n'.join(table_rows_of_clients(clients, keywords)))
 
296
 
 
297
def table_rows_of_clients(clients, keywords):
238
298
    def valuetostring(value, keyword):
239
 
        if type(value) is dbus.Boolean:
 
299
        if isinstance(value, dbus.Boolean):
240
300
            return "Yes" if value else "No"
241
301
        if keyword in ("Timeout", "Interval", "ApprovalDelay",
242
302
                       "ApprovalDuration", "ExtendedTimeout"):
250
310
                      for client in clients)),
251
311
        key=key)
252
312
                             for key in keywords)
253
 
    # Print header line
254
 
    print(format_string.format(**tablewords))
 
313
    # Start with header line
 
314
    rows = [format_string.format(**tablewords)]
255
315
    for client in clients:
256
 
        print(format_string
257
 
              .format(**{key: valuetostring(client[key], key)
258
 
                         for key in keywords}))
 
316
        rows.append(format_string
 
317
                    .format(**{key: valuetostring(client[key], key)
 
318
                               for key in keywords}))
 
319
    return rows
259
320
 
260
321
 
261
322
def has_actions(options):
345
406
    if options.all and not has_actions(options):
346
407
        parser.error("--all requires an action.")
347
408
 
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
 
 
353
409
    try:
354
410
        bus = dbus.SystemBus()
355
411
        mandos_dbus_objc = bus.get_object(busname, server_path)
356
412
    except dbus.exceptions.DBusException:
357
 
        print("Could not connect to Mandos server", file=sys.stderr)
 
413
        log.critical("Could not connect to Mandos server")
358
414
        sys.exit(1)
359
415
 
360
416
    mandos_serv = dbus.Interface(mandos_dbus_objc,
379
435
            os.dup2(stderrcopy, sys.stderr.fileno())
380
436
            os.close(stderrcopy)
381
437
    except dbus.exceptions.DBusException as e:
382
 
        print("Access denied: "
383
 
              "Accessing mandos server through D-Bus: {}".format(e),
384
 
              file=sys.stderr)
 
438
        log.critical("Failed to access Mandos server through D-Bus:"
 
439
                     "\n%s", e)
385
440
        sys.exit(1)
386
441
 
387
442
    # Compile dict of (clients: properties) to process
398
453
                    clients[client_objc] = client
399
454
                    break
400
455
            else:
401
 
                print("Client not found on server: {!r}"
402
 
                      .format(name), file=sys.stderr)
 
456
                log.critical("Client not found on server: %r", name)
403
457
                sys.exit(1)
404
458
 
405
459
    if not has_actions(options) and clients:
493
547
                client.Approve(dbus.Boolean(False),
494
548
                               dbus_interface=client_interface)
495
549
 
 
550
 
 
551
class Test_milliseconds_to_string(unittest.TestCase):
 
552
    def test_all(self):
 
553
        self.assertEqual(milliseconds_to_string(93785000),
 
554
                         "1T02:03:05")
 
555
    def test_no_days(self):
 
556
        self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
 
557
    def test_all_zero(self):
 
558
        self.assertEqual(milliseconds_to_string(0), "00:00:00")
 
559
    def test_no_fractional_seconds(self):
 
560
        self.assertEqual(milliseconds_to_string(400), "00:00:00")
 
561
        self.assertEqual(milliseconds_to_string(900), "00:00:00")
 
562
        self.assertEqual(milliseconds_to_string(1900), "00:00:01")
 
563
 
 
564
class Test_string_to_delta(unittest.TestCase):
 
565
    def test_handles_basic_rfc3339(self):
 
566
        self.assertEqual(string_to_delta("PT2H"),
 
567
                         datetime.timedelta(0, 7200))
 
568
    def test_falls_back_to_pre_1_6_1_with_warning(self):
 
569
        # assertLogs only exists in Python 3.4
 
570
        if hasattr(self, "assertLogs"):
 
571
            with self.assertLogs(log, logging.WARNING):
 
572
                value = string_to_delta("2h")
 
573
        else:
 
574
            value = string_to_delta("2h")
 
575
        self.assertEqual(value, datetime.timedelta(0, 7200))
 
576
 
 
577
 
 
578
def should_only_run_tests():
 
579
    parser = argparse.ArgumentParser(add_help=False)
 
580
    parser.add_argument("--check", action='store_true')
 
581
    args, unknown_args = parser.parse_known_args()
 
582
    run_tests = args.check
 
583
    if run_tests:
 
584
        # Remove --check argument from sys.argv
 
585
        sys.argv[1:] = unknown_args
 
586
    return run_tests
 
587
 
 
588
# Add all tests from doctest strings
 
589
def load_tests(loader, tests, none):
 
590
    import doctest
 
591
    tests.addTests(doctest.DocTestSuite())
 
592
    return tests
496
593
 
497
594
if __name__ == "__main__":
498
 
    main()
 
595
    if should_only_run_tests():
 
596
        # Call using ./tdd-python-script --check [--verbose]
 
597
        unittest.main()
 
598
    else:
 
599
        main()