/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: 2019-12-04 23:46:59 UTC
  • mto: This revision was merged to the branch mainline in revision 396.
  • Revision ID: teddy@recompile.se-20191204234659-m4u7bxnjrns6d3fc
Simplification of Python 3 compatibility code

Apply previously-made change in mandos and mandos-ctl also to
mandos-monitor.

* mandos-monitor: Set "__metaclass__ = type" globally for Python 2,
  and remove inheriting from "object" in all places possible.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
50
50
    __metaclass__ = type
51
51
    str = unicode
52
52
 
53
 
class gi:
54
 
    """Dummy gi module, for the tests"""
55
 
    class repository:
56
 
        class GLib:
57
 
            class Error(Exception):
58
 
                pass
59
 
dbussy = None
60
 
ravel = None
61
 
dbus_python = None
62
 
pydbus = None
63
 
 
64
53
try:
65
 
    import dbussy
66
 
    import ravel
 
54
    import pydbus
 
55
    import gi
 
56
    dbus_python = None
67
57
except ImportError:
68
 
    try:
69
 
        import pydbus
70
 
        import gi
71
 
    except ImportError:
72
 
        import dbus as dbus_python
73
 
 
 
58
    import dbus as dbus_python
 
59
    pydbus = None
 
60
    class gi:
 
61
        """Dummy gi module, for the tests"""
 
62
        class repository:
 
63
            class GLib:
 
64
                class Error(Exception):
 
65
                    pass
74
66
 
75
67
# Show warnings by default
76
68
if not sys.warnoptions:
89
81
 
90
82
locale.setlocale(locale.LC_ALL, "")
91
83
 
92
 
version = "1.8.10"
 
84
version = "1.8.9"
93
85
 
94
86
 
95
87
def main():
104
96
    if options.debug:
105
97
        log.setLevel(logging.DEBUG)
106
98
 
107
 
    if dbussy is not None and ravel is not None:
108
 
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
109
 
    elif pydbus is not None:
 
99
    if pydbus is not None:
110
100
        bus = pydbus_adapter.CachingBus(pydbus)
111
101
    else:
112
102
        bus = dbus_python_adapter.CachingBus(dbus_python)
497
487
                             self.properties_iface, interface, key,
498
488
                             value)
499
489
 
500
 
        def call_method(self, methodname, busname, objectpath,
501
 
                        interface, *args):
502
 
            raise NotImplementedError()
503
 
 
504
490
 
505
491
    class MandosBus(SystemBus):
506
492
        busname_domain = "se.recompile"
696
682
                return new_object
697
683
 
698
684
 
699
 
class dbussy_adapter:
700
 
    class SystemBus(dbus.SystemBus):
701
 
        """Use DBussy"""
702
 
 
703
 
        def __init__(self, dbussy, ravel):
704
 
            self.dbussy = dbussy
705
 
            self.ravel = ravel
706
 
            self.bus = ravel.system_bus()
707
 
 
708
 
        @contextlib.contextmanager
709
 
        def convert_exception(self, exception_class=dbus.Error):
710
 
            try:
711
 
                yield
712
 
            except self.dbussy.DBusError as e:
713
 
                # This does what "raise from" would do
714
 
                exc = exception_class(*e.args)
715
 
                exc.__cause__ = e
716
 
                raise exc
717
 
 
718
 
        def call_method(self, methodname, busname, objectpath,
719
 
                        interface, *args):
720
 
            proxy_object = self.get_object(busname, objectpath)
721
 
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
722
 
                      interface, methodname,
723
 
                      ", ".join(repr(a) for a in args))
724
 
            iface = proxy_object.get_interface(interface)
725
 
            method = getattr(iface, methodname)
726
 
            with self.convert_exception(dbus.Error):
727
 
                value =  method(*args)
728
 
            # DBussy returns values either as an empty list or as a
729
 
            # tuple: (signature, value)
730
 
            if value:
731
 
                return self.type_filter(value[0])
732
 
 
733
 
        def get_object(self, busname, objectpath):
734
 
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
735
 
                      busname, objectpath)
736
 
            with self.convert_exception(dbus.ConnectFailed):
737
 
                return self.bus[busname][objectpath]
738
 
 
739
 
        def type_filter(self, value):
740
 
            """Convert the most bothersome types to Python types"""
741
 
            if isinstance(value, tuple):
742
 
                if (len(value) == 2
743
 
                    and isinstance(value[0],
744
 
                                   self.dbussy.DBUS.Signature)):
745
 
                    return self.type_filter(value[1])
746
 
            elif isinstance(value, self.dbussy.DBUS.ObjectPath):
747
 
                return str(value)
748
 
            # Also recurse into dictionaries
749
 
            elif isinstance(value, dict):
750
 
                return {self.type_filter(key):
751
 
                        self.type_filter(subval)
752
 
                        for key, subval in value.items()}
753
 
            return value
754
 
 
755
 
        def set_property(self, busname, objectpath, interface, key,
756
 
                         value):
757
 
            proxy_object = self.get_object(busname, objectpath)
758
 
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
759
 
                      objectpath, self.properties_iface, interface,
760
 
                      key, value)
761
 
            if key == "Secret":
762
 
                # DBussy wants a Byte Array to be a sequence of
763
 
                # values, not a byte string
764
 
                value = tuple(value)
765
 
            setattr(proxy_object.get_interface(interface), key, value)
766
 
 
767
 
    class MandosBus(SystemBus, dbus.MandosBus):
768
 
        pass
769
 
 
770
 
    class CachingBus(MandosBus):
771
 
        """A caching layer for dbussy_adapter.MandosBus"""
772
 
        def __init__(self, *args, **kwargs):
773
 
            self.object_cache = {}
774
 
            super(dbussy_adapter.CachingBus, self).__init__(*args,
775
 
                                                            **kwargs)
776
 
        def get_object(self, busname, objectpath):
777
 
            try:
778
 
                return self.object_cache[(busname, objectpath)]
779
 
            except KeyError:
780
 
                new_object = super(
781
 
                    dbussy_adapter.CachingBus,
782
 
                    self).get_object(busname, objectpath)
783
 
                self.object_cache[(busname, objectpath)]  = new_object
784
 
                return new_object
785
 
 
786
 
 
787
685
def commands_from_options(options):
788
686
 
789
687
    commands = list(options.commands)
1632
1530
        finally:
1633
1531
            dbus_logger.removeFilter(counting_handler)
1634
1532
 
1635
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1533
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1636
1534
 
1637
1535
        # Make sure the dbus logger was suppressed
1638
1536
        self.assertEqual(0, counting_handler.count)
1775
1673
            self.call_method(bus, "methodname", "busname",
1776
1674
                             "objectpath", "interface")
1777
1675
 
1778
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1676
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1779
1677
 
1780
1678
    def test_get_converts_to_correct_exception(self):
1781
1679
        bus = pydbus_adapter.SystemBus(
1876
1774
        self.assertIs(obj1, obj1b)
1877
1775
 
1878
1776
 
1879
 
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1880
 
 
1881
 
    class dummy_dbussy:
1882
 
        class DBUS:
1883
 
            class ObjectPath(str):
1884
 
                pass
1885
 
        class DBusError(Exception):
1886
 
            pass
1887
 
 
1888
 
    def fake_ravel_func(self, func):
1889
 
        class fake_ravel:
1890
 
            @staticmethod
1891
 
            def system_bus():
1892
 
                class DBusInterfaceProxy:
1893
 
                    @staticmethod
1894
 
                    def methodname(*args):
1895
 
                        return [func(*args)]
1896
 
                class DBusObject:
1897
 
                    @staticmethod
1898
 
                    def get_interface(interface):
1899
 
                        if interface == "interface":
1900
 
                            return DBusInterfaceProxy()
1901
 
                return {"busname": {"objectpath": DBusObject()}}
1902
 
        return fake_ravel
1903
 
 
1904
 
    def call_method(self, bus, methodname, busname, objectpath,
1905
 
                    interface, *args):
1906
 
        with self.assertLogs(log, logging.DEBUG):
1907
 
            return bus.call_method(methodname, busname, objectpath,
1908
 
                                   interface, *args)
1909
 
 
1910
 
    def test_call_method_returns(self):
1911
 
        expected_method_return = Unique()
1912
 
        method_args = (Unique(), Unique())
1913
 
        def func(*args):
1914
 
            self.assertEqual(len(method_args), len(args))
1915
 
            for marg, arg in zip(method_args, args):
1916
 
                self.assertIs(marg, arg)
1917
 
            return expected_method_return
1918
 
        fake_ravel = self.fake_ravel_func(func)
1919
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1920
 
        ret = self.call_method(bus, "methodname", "busname",
1921
 
                               "objectpath", "interface",
1922
 
                               *method_args)
1923
 
        self.assertIs(ret, expected_method_return)
1924
 
 
1925
 
    def test_call_method_filters_objectpath(self):
1926
 
        def func():
1927
 
            return method_return
1928
 
        fake_ravel = self.fake_ravel_func(func)
1929
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1930
 
        method_return = (self.dummy_dbussy.DBUS
1931
 
                         .ObjectPath("objectpath"))
1932
 
        ret = self.call_method(bus, "methodname", "busname",
1933
 
                               "objectpath", "interface")
1934
 
        self.assertEqual("objectpath", ret)
1935
 
        self.assertNotIsInstance(ret,
1936
 
                                 self.dummy_dbussy.DBUS.ObjectPath)
1937
 
 
1938
 
    def test_call_method_filters_objectpaths_in_dict(self):
1939
 
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1940
 
        def func():
1941
 
            return method_return
1942
 
        fake_ravel = self.fake_ravel_func(func)
1943
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1944
 
        method_return = {
1945
 
            ObjectPath("objectpath_key_1"):
1946
 
            ObjectPath("objectpath_value_1"),
1947
 
            ObjectPath("objectpath_key_2"):
1948
 
            ObjectPath("objectpath_value_2"),
1949
 
        }
1950
 
        ret = self.call_method(bus, "methodname", "busname",
1951
 
                               "objectpath", "interface")
1952
 
        expected_method_return = {str(key): str(value)
1953
 
                                  for key, value in
1954
 
                                  method_return.items()}
1955
 
        for key, value in ret.items():
1956
 
            self.assertNotIsInstance(key, ObjectPath)
1957
 
            self.assertNotIsInstance(value, ObjectPath)
1958
 
        self.assertEqual(expected_method_return, ret)
1959
 
        self.assertIsInstance(ret, dict)
1960
 
 
1961
 
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1962
 
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1963
 
        def func():
1964
 
            return method_return
1965
 
        fake_ravel = self.fake_ravel_func(func)
1966
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1967
 
        method_return = {
1968
 
            ObjectPath("key1"): {
1969
 
                ObjectPath("key11"): ObjectPath("value11"),
1970
 
                ObjectPath("key12"): ObjectPath("value12"),
1971
 
            },
1972
 
            ObjectPath("key2"): {
1973
 
                ObjectPath("key21"): ObjectPath("value21"),
1974
 
                ObjectPath("key22"): ObjectPath("value22"),
1975
 
            },
1976
 
        }
1977
 
        ret = self.call_method(bus, "methodname", "busname",
1978
 
                               "objectpath", "interface")
1979
 
        expected_method_return = {
1980
 
            "key1": {"key11": "value11",
1981
 
                     "key12": "value12"},
1982
 
            "key2": {"key21": "value21",
1983
 
                     "key22": "value22"},
1984
 
        }
1985
 
        self.assertEqual(expected_method_return, ret)
1986
 
        for key, value in ret.items():
1987
 
            self.assertIsInstance(value, dict)
1988
 
            self.assertEqual(expected_method_return[key], value)
1989
 
            self.assertNotIsInstance(key, ObjectPath)
1990
 
            for inner_key, inner_value in value.items():
1991
 
                self.assertIsInstance(value, dict)
1992
 
                self.assertEqual(
1993
 
                    expected_method_return[key][inner_key],
1994
 
                    inner_value)
1995
 
                self.assertNotIsInstance(key, ObjectPath)
1996
 
 
1997
 
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
1998
 
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1999
 
        def func():
2000
 
            return method_return
2001
 
        fake_ravel = self.fake_ravel_func(func)
2002
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2003
 
        method_return = {
2004
 
            ObjectPath("key1"): {
2005
 
                ObjectPath("key2"): {
2006
 
                    ObjectPath("key3"): ObjectPath("value"),
2007
 
                },
2008
 
            },
2009
 
        }
2010
 
        ret = self.call_method(bus, "methodname", "busname",
2011
 
                               "objectpath", "interface")
2012
 
        expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2013
 
        self.assertEqual(expected_method_return, ret)
2014
 
        self.assertIsInstance(ret, dict)
2015
 
        self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2016
 
        self.assertIsInstance(ret["key1"], dict)
2017
 
        self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2018
 
                                 ObjectPath)
2019
 
        self.assertIsInstance(ret["key1"]["key2"], dict)
2020
 
        self.assertNotIsInstance(
2021
 
            next(iter(ret["key1"]["key2"].keys())),
2022
 
            ObjectPath)
2023
 
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
2024
 
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2025
 
                                 self.dummy_dbussy.DBUS.ObjectPath)
2026
 
 
2027
 
    def test_call_method_handles_exception(self):
2028
 
        def func():
2029
 
            raise self.dummy_dbussy.DBusError()
2030
 
 
2031
 
        fake_ravel = self.fake_ravel_func(func)
2032
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2033
 
 
2034
 
        with self.assertRaises(dbus.Error) as e:
2035
 
            self.call_method(bus, "methodname", "busname",
2036
 
                             "objectpath", "interface")
2037
 
 
2038
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2039
 
 
2040
 
    def test_get_object_converts_to_correct_exception(self):
2041
 
        class fake_ravel_raises_exception_on_connect:
2042
 
            @staticmethod
2043
 
            def system_bus():
2044
 
                class Bus:
2045
 
                    @staticmethod
2046
 
                    def __getitem__(key):
2047
 
                        if key == "objectpath":
2048
 
                            raise self.dummy_dbussy.DBusError()
2049
 
                        raise Exception(key)
2050
 
                return {"busname": Bus()}
2051
 
        def func():
2052
 
            raise self.dummy_dbussy.DBusError()
2053
 
        bus = dbussy_adapter.SystemBus(
2054
 
            self.dummy_dbussy,
2055
 
            fake_ravel_raises_exception_on_connect)
2056
 
        with self.assertRaises(dbus.ConnectFailed):
2057
 
            self.call_method(bus, "methodname", "busname",
2058
 
                             "objectpath", "interface")
2059
 
 
2060
 
 
2061
1777
class Test_commands_from_options(unittest.TestCase):
2062
1778
 
2063
1779
    def setUp(self):