/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-31 09:13:31 UTC
  • Revision ID: teddy@recompile.se-20190331091331-avb4odg8uz1yb5tm
mandos-ctl: Add support for D-Bus module "pydbus"

* mandos-ctl: Try to import pydbus and gi modules.
  (main): Use pydbus if available, fall back to dbus-python.
  (pydbus_adapter): New.
  (Test_pydbus_adapter_SystemBus): - '' -
  (Test_pydbus_adapter_CachingBus): - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
46
46
import tempfile
47
47
import contextlib
48
48
 
49
 
import dbus as dbus_python
 
49
try:
 
50
    import pydbus
 
51
    import gi
 
52
    dbus_python = None
 
53
except ImportError:
 
54
    import dbus as dbus_python
 
55
    pydbus = None
 
56
    class gi(object):
 
57
        """Dummy gi module, for the tests"""
 
58
        class repository(object):
 
59
            class GLib(object):
 
60
                class Error(Exception):
 
61
                    pass
50
62
 
51
63
# Show warnings by default
52
64
if not sys.warnoptions:
81
93
    if options.debug:
82
94
        log.setLevel(logging.DEBUG)
83
95
 
84
 
    bus = dbus_python_adapter.CachingBus(dbus_python)
 
96
    if pydbus is not None:
 
97
        bus = pydbus_adapter.CachingBus(pydbus)
 
98
    else:
 
99
        bus = dbus_python_adapter.CachingBus(dbus_python)
85
100
 
86
101
    try:
87
102
        all_clients = bus.get_clients_and_properties()
592
607
                return new_object
593
608
 
594
609
 
 
610
class pydbus_adapter(object):
 
611
    class SystemBus(dbus.MandosBus):
 
612
        def __init__(self, module=pydbus):
 
613
            self.pydbus = module
 
614
            self.bus = self.pydbus.SystemBus()
 
615
 
 
616
        @contextlib.contextmanager
 
617
        def convert_exception(self, exception_class=dbus.Error):
 
618
            try:
 
619
                yield
 
620
            except gi.repository.GLib.Error as e:
 
621
                # This does what "raise from" would do
 
622
                exc = exception_class(*e.args)
 
623
                exc.__cause__ = e
 
624
                raise exc
 
625
 
 
626
        def call_method(self, methodname, busname, objectpath,
 
627
                        interface, *args):
 
628
            proxy_object = self.get(busname, objectpath)
 
629
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
630
                      interface, methodname,
 
631
                      ", ".join(repr(a) for a in args))
 
632
            method = getattr(proxy_object[interface], methodname)
 
633
            with self.convert_exception():
 
634
                return method(*args)
 
635
 
 
636
        def get(self, busname, objectpath):
 
637
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
638
                      busname, objectpath)
 
639
            with self.convert_exception(dbus.ConnectFailed):
 
640
                if sys.version_info.major <= 2:
 
641
                    with warnings.catch_warnings():
 
642
                        warnings.filterwarnings(
 
643
                            "ignore", "", DeprecationWarning,
 
644
                            r"^xml\.etree\.ElementTree$")
 
645
                        return self.bus.get(busname, objectpath)
 
646
                else:
 
647
                    return self.bus.get(busname, objectpath)
 
648
 
 
649
        def set_property(self, busname, objectpath, interface, key,
 
650
                         value):
 
651
            proxy_object = self.get(busname, objectpath)
 
652
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
653
                      objectpath, self.properties_iface, interface,
 
654
                      key, value)
 
655
            setattr(proxy_object[interface], key, value)
 
656
 
 
657
    class CachingBus(SystemBus):
 
658
        """A caching layer for pydbus_adapter.SystemBus"""
 
659
        def __init__(self, *args, **kwargs):
 
660
            self.object_cache = {}
 
661
            super(pydbus_adapter.CachingBus,
 
662
                  self).__init__(*args, **kwargs)
 
663
        def get(self, busname, objectpath):
 
664
            try:
 
665
                return self.object_cache[(busname, objectpath)]
 
666
            except KeyError:
 
667
                new_object = (super(pydbus_adapter.CachingBus, self)
 
668
                              .get(busname, objectpath))
 
669
                self.object_cache[(busname, objectpath)]  = new_object
 
670
                return new_object
 
671
 
 
672
 
595
673
def commands_from_options(options):
596
674
 
597
675
    commands = list(options.commands)
1508
1586
        self.assertIs(obj1, obj1b)
1509
1587
 
1510
1588
 
 
1589
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
 
1590
 
 
1591
    def Stub_pydbus_func(self, func):
 
1592
        class stub_pydbus(object):
 
1593
            """stub pydbus module"""
 
1594
            class SystemBus(object):
 
1595
                @staticmethod
 
1596
                def get(busname, objectpath):
 
1597
                    DBusObject = collections.namedtuple(
 
1598
                        "DBusObject", ("methodname",))
 
1599
                    return {"interface":
 
1600
                            DBusObject(methodname=func)}
 
1601
        return stub_pydbus
 
1602
 
 
1603
    def call_method(self, bus, methodname, busname, objectpath,
 
1604
                    interface, *args):
 
1605
        with self.assertLogs(log, logging.DEBUG):
 
1606
            return bus.call_method(methodname, busname, objectpath,
 
1607
                                   interface, *args)
 
1608
 
 
1609
    def test_call_method_returns(self):
 
1610
        expected_method_return = Unique()
 
1611
        method_args = (Unique(), Unique())
 
1612
        def func(*args):
 
1613
            self.assertEqual(len(method_args), len(args))
 
1614
            for marg, arg in zip(method_args, args):
 
1615
                self.assertIs(marg, arg)
 
1616
            return expected_method_return
 
1617
        stub_pydbus = self.Stub_pydbus_func(func)
 
1618
        bus = pydbus_adapter.SystemBus(stub_pydbus)
 
1619
        ret = self.call_method(bus, "methodname", "busname",
 
1620
                               "objectpath", "interface",
 
1621
                               *method_args)
 
1622
        self.assertIs(ret, expected_method_return)
 
1623
 
 
1624
    def test_call_method_handles_exception(self):
 
1625
        dbus_logger = logging.getLogger("dbus.proxies")
 
1626
 
 
1627
        def func():
 
1628
            raise gi.repository.GLib.Error()
 
1629
 
 
1630
        stub_pydbus = self.Stub_pydbus_func(func)
 
1631
        bus = pydbus_adapter.SystemBus(stub_pydbus)
 
1632
 
 
1633
        with self.assertRaises(dbus.Error) as e:
 
1634
            self.call_method(bus, "methodname", "busname",
 
1635
                             "objectpath", "interface")
 
1636
 
 
1637
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1638
 
 
1639
    def test_get_converts_to_correct_exception(self):
 
1640
        bus = pydbus_adapter.SystemBus(
 
1641
            self.fake_pydbus_raises_exception_on_connect)
 
1642
        with self.assertRaises(dbus.ConnectFailed):
 
1643
            self.call_method(bus, "methodname", "busname",
 
1644
                             "objectpath", "interface")
 
1645
 
 
1646
    class fake_pydbus_raises_exception_on_connect(object):
 
1647
        """fake dbus-python module"""
 
1648
        @classmethod
 
1649
        def SystemBus(cls):
 
1650
            def get(busname, objectpath):
 
1651
                raise gi.repository.GLib.Error()
 
1652
            Bus = collections.namedtuple("Bus", ["get"])
 
1653
            return Bus(get=get)
 
1654
 
 
1655
    def test_set_property_uses_setattr(self):
 
1656
        class Object(object):
 
1657
            pass
 
1658
        obj = Object()
 
1659
        class pydbus_spy(object):
 
1660
            class SystemBus(object):
 
1661
                @staticmethod
 
1662
                def get(busname, objectpath):
 
1663
                    return {"interface": obj}
 
1664
        bus = pydbus_adapter.SystemBus(pydbus_spy)
 
1665
        value = Unique()
 
1666
        bus.set_property("busname", "objectpath", "interface", "key",
 
1667
                         value)
 
1668
        self.assertIs(value, obj.key)
 
1669
 
 
1670
    def test_get_suppresses_xml_deprecation_warning(self):
 
1671
        if sys.version_info.major >= 3:
 
1672
            return
 
1673
        class stub_pydbus_get(object):
 
1674
            class SystemBus(object):
 
1675
                @staticmethod
 
1676
                def get(busname, objectpath):
 
1677
                    warnings.warn_explicit(
 
1678
                        "deprecated", DeprecationWarning,
 
1679
                        "xml.etree.ElementTree", 0)
 
1680
        bus = pydbus_adapter.SystemBus(stub_pydbus_get)
 
1681
        with warnings.catch_warnings(record=True) as w:
 
1682
            warnings.simplefilter("always")
 
1683
            bus.get("busname", "objectpath")
 
1684
            self.assertEqual(0, len(w))
 
1685
 
 
1686
 
 
1687
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
 
1688
    class stub_pydbus(object):
 
1689
        """stub pydbus module"""
 
1690
        class SystemBus(object):
 
1691
            @staticmethod
 
1692
            def get(busname, objectpath):
 
1693
                return Unique()
 
1694
 
 
1695
    def setUp(self):
 
1696
        self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
 
1697
 
 
1698
    def test_returns_distinct_objectpaths(self):
 
1699
        obj1 = self.bus.get("busname", "objectpath1")
 
1700
        self.assertIsInstance(obj1, Unique)
 
1701
        obj2 = self.bus.get("busname", "objectpath2")
 
1702
        self.assertIsInstance(obj2, Unique)
 
1703
        self.assertIsNot(obj1, obj2)
 
1704
 
 
1705
    def test_returns_distinct_busnames(self):
 
1706
        obj1 = self.bus.get("busname1", "objectpath")
 
1707
        self.assertIsInstance(obj1, Unique)
 
1708
        obj2 = self.bus.get("busname2", "objectpath")
 
1709
        self.assertIsInstance(obj2, Unique)
 
1710
        self.assertIsNot(obj1, obj2)
 
1711
 
 
1712
    def test_returns_distinct_both(self):
 
1713
        obj1 = self.bus.get("busname1", "objectpath")
 
1714
        self.assertIsInstance(obj1, Unique)
 
1715
        obj2 = self.bus.get("busname2", "objectpath")
 
1716
        self.assertIsInstance(obj2, Unique)
 
1717
        self.assertIsNot(obj1, obj2)
 
1718
 
 
1719
    def test_returns_same(self):
 
1720
        obj1 = self.bus.get("busname", "objectpath")
 
1721
        self.assertIsInstance(obj1, Unique)
 
1722
        obj2 = self.bus.get("busname", "objectpath")
 
1723
        self.assertIsInstance(obj2, Unique)
 
1724
        self.assertIs(obj1, obj2)
 
1725
 
 
1726
    def test_returns_same_old(self):
 
1727
        obj1 = self.bus.get("busname1", "objectpath1")
 
1728
        self.assertIsInstance(obj1, Unique)
 
1729
        obj2 = self.bus.get("busname2", "objectpath2")
 
1730
        self.assertIsInstance(obj2, Unique)
 
1731
        obj1b = self.bus.get("busname1", "objectpath1")
 
1732
        self.assertIsInstance(obj1b, Unique)
 
1733
        self.assertIsNot(obj1, obj2)
 
1734
        self.assertIsNot(obj2, obj1b)
 
1735
        self.assertIs(obj1, obj1b)
 
1736
 
 
1737
 
1511
1738
class Test_commands_from_options(unittest.TestCase):
1512
1739
 
1513
1740
    def setUp(self):