/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 at recompile
  • Date: 2020-01-12 01:42:09 UTC
  • mto: This revision was merged to the branch mainline in revision 396.
  • Revision ID: teddy@recompile.se-20200112014209-ktr3acloxzbmhbnt
mandos-ctl: Add DBussy support

Add support in mandos-ctl for the "DBussy" Python D-Bus module.  Use
it by default, if it is available.

* mandos-ctl: Try to import the "dbussy" and its high-level module
  "ravel".
  (main): Use DBussy if import succeeded.
  (dbussy_adapter): New.
  (Test_dbussy_adapter_SystemBus): New test class.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
1
#!/usr/bin/python3 -bbI
2
 
# -*- coding: utf-8; lexical-binding: t -*-
3
 
#
4
 
# Mandos Control - Control or query the Mandos server
5
 
#
6
 
# Copyright © 2008-2022 Teddy Hogeborn
7
 
# Copyright © 2008-2022 Björn Påhlsson
 
2
# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-
 
3
#
 
4
# Mandos Monitor - Control and monitor the Mandos server
 
5
#
 
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
#
23
23
#
24
24
# Contact the authors at <mandos@recompile.se>.
25
25
#
 
26
 
26
27
from __future__ import (division, absolute_import, print_function,
27
28
                        unicode_literals)
28
29
 
32
33
    pass
33
34
 
34
35
import sys
35
 
import unittest
36
36
import argparse
37
 
import logging
38
 
import os
39
37
import locale
40
38
import datetime
41
39
import re
 
40
import os
42
41
import collections
43
42
import json
 
43
import unittest
 
44
import logging
44
45
import io
45
46
import tempfile
46
47
import contextlib
48
49
if sys.version_info.major == 2:
49
50
    __metaclass__ = type
50
51
    str = unicode
51
 
    input = raw_input
52
52
 
53
53
class gi:
54
54
    """Dummy gi module, for the tests"""
77
77
    import warnings
78
78
    warnings.simplefilter("default")
79
79
 
80
 
log = logging.getLogger(os.path.basename(sys.argv[0]))
 
80
log = logging.getLogger(sys.argv[0])
81
81
logging.basicConfig(level="INFO", # Show info level messages
82
82
                    format="%(message)s") # Show basic log messages
83
83
 
89
89
 
90
90
locale.setlocale(locale.LC_ALL, "")
91
91
 
92
 
version = "1.8.15"
 
92
version = "1.8.9"
93
93
 
94
94
 
95
95
def main():
102
102
    clientnames = options.client
103
103
 
104
104
    if options.debug:
105
 
        logging.getLogger("").setLevel(logging.DEBUG)
 
105
        log.setLevel(logging.DEBUG)
106
106
 
107
107
    if dbussy is not None and ravel is not None:
108
108
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
256
256
        return rfc3339_duration_to_delta(interval)
257
257
    except ValueError as e:
258
258
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
259
 
                    " ".join(e.args))
 
259
                    ' '.join(e.args))
260
260
    return parse_pre_1_6_1_interval(interval)
261
261
 
262
262
 
395
395
    """Parse an interval string as documented by Mandos before 1.6.1,
396
396
    and return a datetime.timedelta
397
397
 
398
 
    >>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)
399
 
    True
400
 
    >>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)
401
 
    True
402
 
    >>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)
403
 
    True
404
 
    >>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)
405
 
    True
406
 
    >>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)
407
 
    True
408
 
    >>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)
409
 
    True
410
 
    >>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)
 
398
    >>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)
 
399
    True
 
400
    >>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)
 
401
    True
 
402
    >>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)
 
403
    True
 
404
    >>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)
 
405
    True
 
406
    >>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)
 
407
    True
 
408
    >>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)
 
409
    True
 
410
    >>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)
411
411
    True
412
412
    >>> # Ignore unknown characters, allow any order and repetitions
413
 
    >>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") == datetime.timedelta(2, 480, 18000)
 
413
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
414
414
    True
415
415
 
416
416
    """
497
497
                             self.properties_iface, interface, key,
498
498
                             value)
499
499
 
500
 
        def call_method(self, methodname, busname, objectpath,
501
 
                        interface, *args):
502
 
            raise NotImplementedError()
503
 
 
504
 
 
505
500
    class MandosBus(SystemBus):
506
501
        busname_domain = "se.recompile"
507
502
        busname = busname_domain + ".Mandos"
726
721
            with self.convert_exception(dbus.Error):
727
722
                value =  method(*args)
728
723
            # DBussy returns values either as an empty list or as a
729
 
            # list of one element with the return value
 
724
            # tuple: (signature, value)
730
725
            if value:
731
726
                return self.type_filter(value[0])
732
727
 
738
733
 
739
734
        def type_filter(self, value):
740
735
            """Convert the most bothersome types to Python types"""
741
 
            # A D-Bus Variant value is represented as the Python type
742
 
            # Tuple[dbussy.DBUS.Signature, Any]
743
736
            if isinstance(value, tuple):
744
737
                if (len(value) == 2
745
738
                    and isinstance(value[0],
876
869
                    {key: properties[key]
877
870
                     for key in self.all_keywords}
878
871
                    for properties in clients.values()}
879
 
            print(json.dumps(data, indent=4, separators=(",", ": ")))
 
872
            print(json.dumps(data, indent=4, separators=(',', ': ')))
880
873
 
881
874
 
882
875
    class PrintTable(Output):
1634
1627
        finally:
1635
1628
            dbus_logger.removeFilter(counting_handler)
1636
1629
 
1637
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1630
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1638
1631
 
1639
1632
        # Make sure the dbus logger was suppressed
1640
1633
        self.assertEqual(0, counting_handler.count)
1777
1770
            self.call_method(bus, "methodname", "busname",
1778
1771
                             "objectpath", "interface")
1779
1772
 
1780
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1773
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1781
1774
 
1782
1775
    def test_get_converts_to_correct_exception(self):
1783
1776
        bus = pydbus_adapter.SystemBus(
2439
2432
        busname = "se.recompile.Mandos"
2440
2433
        client_interface = "se.recompile.Mandos.Client"
2441
2434
        command.Approve().run(self.bus.clients, self.bus)
2442
 
        self.assertTrue(self.bus.clients)
2443
2435
        for clientpath in self.bus.clients:
2444
2436
            self.assertIn(("Approve", busname, clientpath,
2445
2437
                           client_interface, (True,)), self.bus.calls)
2448
2440
        busname = "se.recompile.Mandos"
2449
2441
        client_interface = "se.recompile.Mandos.Client"
2450
2442
        command.Deny().run(self.bus.clients, self.bus)
2451
 
        self.assertTrue(self.bus.clients)
2452
2443
        for clientpath in self.bus.clients:
2453
2444
            self.assertIn(("Approve", busname, clientpath,
2454
2445
                           client_interface, (False,)),
2455
2446
                          self.bus.calls)
2456
2447
 
2457
2448
    def test_Remove(self):
2458
 
        busname = "se.recompile.Mandos"
2459
 
        server_path = "/"
2460
 
        server_interface = "se.recompile.Mandos"
2461
 
        orig_clients = self.bus.clients.copy()
2462
2449
        command.Remove().run(self.bus.clients, self.bus)
2463
 
        self.assertFalse(self.bus.clients)
2464
 
        for clientpath in orig_clients:
2465
 
            self.assertIn(("RemoveClient", busname,
2466
 
                           server_path, server_interface,
 
2450
        for clientpath in self.bus.clients:
 
2451
            self.assertIn(("RemoveClient", dbus_busname,
 
2452
                           dbus_server_path, dbus_server_interface,
2467
2453
                           (clientpath,)), self.bus.calls)
2468
2454
 
2469
2455
    expected_json = {
2671
2657
        else:
2672
2658
            cmd_args = [() for x in range(len(self.values_to_get))]
2673
2659
            values_to_get = self.values_to_get
2674
 
        self.assertTrue(values_to_get)
2675
2660
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2676
2661
            for clientpath in self.bus.clients:
2677
2662
                self.bus.clients[clientpath][self.propname] = (
2678
2663
                    Unique())
2679
2664
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
2680
 
            self.assertTrue(self.bus.clients)
2681
2665
            for clientpath in self.bus.clients:
2682
2666
                value = (self.bus.clients[clientpath]
2683
2667
                         [self.propname])
2742
2726
class TestSetSecretCmd(TestPropertySetterCmd):
2743
2727
    command = command.SetSecret
2744
2728
    propname = "Secret"
2745
 
    def __init__(self, *args, **kwargs):
2746
 
        self.values_to_set = [io.BytesIO(b""),
2747
 
                              io.BytesIO(b"secret\0xyzzy\nbar")]
2748
 
        self.values_to_get = [f.getvalue() for f in
2749
 
                              self.values_to_set]
2750
 
        super(TestSetSecretCmd, self).__init__(*args, **kwargs)
 
2729
    values_to_set = [io.BytesIO(b""),
 
2730
                     io.BytesIO(b"secret\0xyzzy\nbar")]
 
2731
    values_to_get = [f.getvalue() for f in values_to_set]
2751
2732
 
2752
2733
 
2753
2734
class TestSetTimeoutCmd(TestPropertySetterCmd):
2806
2787
 
2807
2788
 
2808
2789
 
2809
 
def parse_test_args():
2810
 
    # type: () -> argparse.Namespace
 
2790
def should_only_run_tests():
2811
2791
    parser = argparse.ArgumentParser(add_help=False)
2812
 
    parser.add_argument("--check", action="store_true")
2813
 
    parser.add_argument("--prefix", )
 
2792
    parser.add_argument("--check", action='store_true')
2814
2793
    args, unknown_args = parser.parse_known_args()
2815
 
    if args.check:
2816
 
        # Remove test options from sys.argv
 
2794
    run_tests = args.check
 
2795
    if run_tests:
 
2796
        # Remove --check argument from sys.argv
2817
2797
        sys.argv[1:] = unknown_args
2818
 
    return args
 
2798
    return run_tests
2819
2799
 
2820
2800
# Add all tests from doctest strings
2821
2801
def load_tests(loader, tests, none):
2824
2804
    return tests
2825
2805
 
2826
2806
if __name__ == "__main__":
2827
 
    options = parse_test_args()
2828
2807
    try:
2829
 
        if options.check:
2830
 
            extra_test_prefix = options.prefix
2831
 
            if extra_test_prefix is not None:
2832
 
                if not (unittest.main(argv=[""], exit=False)
2833
 
                        .result.wasSuccessful()):
2834
 
                    sys.exit(1)
2835
 
                class ExtraTestLoader(unittest.TestLoader):
2836
 
                    testMethodPrefix = extra_test_prefix
2837
 
                # Call using ./scriptname --check [--verbose]
2838
 
                unittest.main(argv=[""], testLoader=ExtraTestLoader())
2839
 
            else:
2840
 
                unittest.main(argv=[""])
 
2808
        if should_only_run_tests():
 
2809
            # Call using ./tdd-python-script --check [--verbose]
 
2810
            unittest.main()
2841
2811
        else:
2842
2812
            main()
2843
2813
    finally:
2844
2814
        logging.shutdown()
2845
 
 
2846
 
# Local Variables:
2847
 
# run-tests:
2848
 
# (lambda (&optional extra)
2849
 
#   (if (not (funcall run-tests-in-test-buffer default-directory
2850
 
#             extra))
2851
 
#       (funcall show-test-buffer-in-test-window)
2852
 
#     (funcall remove-test-window)
2853
 
#     (if extra (message "Extra tests run successfully!"))))
2854
 
# run-tests-in-test-buffer:
2855
 
# (lambda (dir &optional extra)
2856
 
#   (with-current-buffer (get-buffer-create "*Test*")
2857
 
#     (setq buffer-read-only nil
2858
 
#           default-directory dir)
2859
 
#     (erase-buffer)
2860
 
#     (compilation-mode))
2861
 
#   (let ((process-result
2862
 
#          (let ((inhibit-read-only t))
2863
 
#            (process-file-shell-command
2864
 
#             (funcall get-command-line extra) nil "*Test*"))))
2865
 
#     (and (numberp process-result)
2866
 
#          (= process-result 0))))
2867
 
# get-command-line:
2868
 
# (lambda (&optional extra)
2869
 
#   (let ((quoted-script
2870
 
#          (shell-quote-argument (funcall get-script-name))))
2871
 
#     (format
2872
 
#      (concat "%s --check" (if extra " --prefix=atest" ""))
2873
 
#      quoted-script)))
2874
 
# get-script-name:
2875
 
# (lambda ()
2876
 
#   (if (fboundp 'file-local-name)
2877
 
#       (file-local-name (buffer-file-name))
2878
 
#     (or (file-remote-p (buffer-file-name) 'localname)
2879
 
#         (buffer-file-name))))
2880
 
# remove-test-window:
2881
 
# (lambda ()
2882
 
#   (let ((test-window (get-buffer-window "*Test*")))
2883
 
#     (if test-window (delete-window test-window))))
2884
 
# show-test-buffer-in-test-window:
2885
 
# (lambda ()
2886
 
#   (when (not (get-buffer-window-list "*Test*"))
2887
 
#     (setq next-error-last-buffer (get-buffer "*Test*"))
2888
 
#     (let* ((side (if (>= (window-width) 146) 'right 'bottom))
2889
 
#            (display-buffer-overriding-action
2890
 
#             `((display-buffer-in-side-window) (side . ,side)
2891
 
#               (window-height . fit-window-to-buffer)
2892
 
#               (window-width . fit-window-to-buffer))))
2893
 
#       (display-buffer "*Test*"))))
2894
 
# eval:
2895
 
# (progn
2896
 
#   (let* ((run-extra-tests (lambda () (interactive)
2897
 
#                             (funcall run-tests t)))
2898
 
#          (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
2899
 
#          (outer-keymap `(keymap (3 . ,inner-keymap))))     ; C-c
2900
 
#     (setq minor-mode-overriding-map-alist
2901
 
#           (cons `(run-tests . ,outer-keymap)
2902
 
#                 minor-mode-overriding-map-alist)))
2903
 
#   (add-hook 'after-save-hook run-tests 90 t))
2904
 
# End: