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
- browse files at revision 237.7.761
- compare with another revision
- download diff
- download tarball
- view history from revision 237.7.761
- Committer: teddy at recompile
- Date: 2020-04-05 21:30:59 UTC
- mto: This revision was merged to the branch mainline in revision 398.
- Revision ID: teddy@recompile.se-20200405213059-fb2a61ckqynrmatk
Fix file descriptor leak in mandos-client
When the local network has Mandos servers announcing themselves using
real, globally reachable, IPv6 addresses (i.e. not link-local
addresses), but there is no router on the local network providing IPv6
RA (Router Advertisement) packets, the client cannot reach the server
by normal means, since the client only has a link-local IPv6 address,
and has no usable route to reach the server's global IPv6 address.
(This is not a common situation, and usually only happens when the
router itself reboots and runs a Mandos client, since it cannot then
give RA packets to itself.) The client code has a solution for
this, which consists of adding a temporary local route to reach the
address of the server during communication, and removing this
temporary route afterwards.
This solution with a temporary route works, but has a file descriptor
leak; it leaks one file descriptor for each addition and for each
removal of a route. If one server requiring an added route is present
on the network, but no servers gives a password, making the client
retry after the default ten seconds, and we furthermore assume a
default 1024 open files limit, the client runs out of file descriptors
after about 90 minutes, after which time the client process will be
useless and fail to retrieve any passwords, necessitating manual
password entry via the keyboard.
Fix this by eliminating the file descriptor leak in the client.
* plugins.d/mandos-client.c (add_delete_local_route): Do
close(devnull) also in parent process, also if fork() fails, and on
any failure in child process.
When the local network has Mandos servers announcing themselves using
real, globally reachable, IPv6 addresses (i.e. not link-local
addresses), but there is no router on the local network providing IPv6
RA (Router Advertisement) packets, the client cannot reach the server
by normal means, since the client only has a link-local IPv6 address,
and has no usable route to reach the server's global IPv6 address.
(This is not a common situation, and usually only happens when the
router itself reboots and runs a Mandos client, since it cannot then
give RA packets to itself.) The client code has a solution for
this, which consists of adding a temporary local route to reach the
address of the server during communication, and removing this
temporary route afterwards.
This solution with a temporary route works, but has a file descriptor
leak; it leaks one file descriptor for each addition and for each
removal of a route. If one server requiring an added route is present
on the network, but no servers gives a password, making the client
retry after the default ten seconds, and we furthermore assume a
default 1024 open files limit, the client runs out of file descriptors
after about 90 minutes, after which time the client process will be
useless and fail to retrieve any passwords, necessitating manual
password entry via the keyboard.
Fix this by eliminating the file descriptor leak in the client.
* plugins.d/mandos-client.c (add_delete_local_route): Do
close(devnull) also in parent process, also if fork() fails, and on
any failure in child process.
- files added:
- debian/po/POTFILES.in
- debian/tests
- dracut-module
- files modified:
- .bzrignore
added
removed
mandos-ctl
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*")))); -*-
1
#!/usr/bin/python3 -bbI
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
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
46
46
import tempfile
47
47
import contextlib
48
48
49
if sys.version_info.major == 2:
50
__metaclass__ = type
51
str = unicode
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
49
64
try:
50
import pydbus
51
import gi
52
dbus_python = None
65
import dbussy
66
import ravel
53
67
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
68
try:
69
import pydbus
70
import gi
71
except ImportError:
72
import dbus as dbus_python
73
62
74
63
75
# Show warnings by default
64
76
if not sys.warnoptions:
72
84
logging.captureWarnings(True) # Show warnings via the logging system
73
85
74
86
if sys.version_info.major == 2:
75
str = unicode
76
87
import StringIO
77
88
io.StringIO = StringIO.StringIO
78
89
79
90
locale.setlocale(locale.LC_ALL, "")
80
91
81
version = "1.8.3"
92
version = "1.8.10"
82
93
83
94
84
95
def main():
93
104
if options.debug:
94
105
log.setLevel(logging.DEBUG)
95
106
96
if pydbus is not None:
107
if dbussy is not None and ravel is not None:
108
bus = dbussy_adapter.CachingBus(dbussy, ravel)
109
elif pydbus is not None:
97
110
bus = pydbus_adapter.CachingBus(pydbus)
98
111
else:
99
112
bus = dbus_python_adapter.CachingBus(dbus_python)
250
263
def rfc3339_duration_to_delta(duration):
251
264
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
252
265
253
>>> rfc3339_duration_to_delta("P7D")
254
datetime.timedelta(7)
255
>>> rfc3339_duration_to_delta("PT60S")
256
datetime.timedelta(0, 60)
257
>>> rfc3339_duration_to_delta("PT60M")
258
datetime.timedelta(0, 3600)
259
>>> rfc3339_duration_to_delta("P60M")
260
datetime.timedelta(1680)
261
>>> rfc3339_duration_to_delta("PT24H")
262
datetime.timedelta(1)
263
>>> rfc3339_duration_to_delta("P1W")
264
datetime.timedelta(7)
265
>>> rfc3339_duration_to_delta("PT5M30S")
266
datetime.timedelta(0, 330)
267
>>> rfc3339_duration_to_delta("P1DT3M20S")
268
datetime.timedelta(1, 200)
266
>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
267
True
268
>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
269
True
270
>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
271
True
272
>>> # 60 months
273
>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
274
True
275
>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
276
True
277
>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
278
True
279
>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
280
True
281
>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
282
True
269
283
>>> # Can not be empty:
270
284
>>> rfc3339_duration_to_delta("")
271
285
Traceback (most recent call last):
381
395
"""Parse an interval string as documented by Mandos before 1.6.1,
382
396
and return a datetime.timedelta
383
397
384
>>> parse_pre_1_6_1_interval('7d')
385
datetime.timedelta(7)
386
>>> parse_pre_1_6_1_interval('60s')
387
datetime.timedelta(0, 60)
388
>>> parse_pre_1_6_1_interval('60m')
389
datetime.timedelta(0, 3600)
390
>>> parse_pre_1_6_1_interval('24h')
391
datetime.timedelta(1)
392
>>> parse_pre_1_6_1_interval('1w')
393
datetime.timedelta(7)
394
>>> parse_pre_1_6_1_interval('5m 30s')
395
datetime.timedelta(0, 330)
396
>>> parse_pre_1_6_1_interval('')
397
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
True
398
412
>>> # Ignore unknown characters, allow any order and repetitions
399
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
datetime.timedelta(2, 480, 18000)
413
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
414
True
401
415
402
416
"""
403
417
466
480
parser.error("--remove can only be combined with --deny")
467
481
468
482
469
class dbus(object):
483
class dbus:
470
484
471
class SystemBus(object):
485
class SystemBus:
472
486
473
487
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
474
488
def get_managed_objects(self, busname, objectpath):
483
497
self.properties_iface, interface, key,
484
498
value)
485
499
500
def call_method(self, methodname, busname, objectpath,
501
interface, *args):
502
raise NotImplementedError()
503
486
504
487
505
class MandosBus(SystemBus):
488
506
busname_domain = "se.recompile"
520
538
pass
521
539
522
540
523
class dbus_python_adapter(object):
541
class dbus_python_adapter:
524
542
525
543
class SystemBus(dbus.MandosBus):
526
544
"""Use dbus-python"""
571
589
for key, subval in value.items()}
572
590
return value
573
591
592
def set_client_property(self, objectpath, key, value):
593
if key == "Secret":
594
if not isinstance(value, bytes):
595
value = value.encode("utf-8")
596
value = self.dbus_python.ByteArray(value)
597
return self.set_property(self.busname, objectpath,
598
self.client_interface, key,
599
value)
574
600
575
class SilenceLogger(object):
601
class SilenceLogger:
576
602
"Simple context manager to silence a particular logger"
577
603
def __init__(self, loggername):
578
604
self.logger = logging.getLogger(loggername)
607
633
return new_object
608
634
609
635
610
class pydbus_adapter(object):
636
class pydbus_adapter:
611
637
class SystemBus(dbus.MandosBus):
612
638
def __init__(self, module=pydbus):
613
639
self.pydbus = module
670
696
return new_object
671
697
672
698
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
673
787
def commands_from_options(options):
674
788
675
789
commands = list(options.commands)
703
817
return commands
704
818
705
819
706
class command(object):
820
class command:
707
821
"""A namespace for command classes"""
708
822
709
class Base(object):
823
class Base:
710
824
"""Abstract base class for commands"""
711
825
def run(self, clients, bus=None):
712
826
"""Normal commands should implement run_on_one_client(),
775
889
keywords = self.all_keywords
776
890
print(self.TableOfClients(clients.values(), keywords))
777
891
778
class TableOfClients(object):
892
class TableOfClients:
779
893
tableheaders = {
780
894
"Name": "Name",
781
895
"Enabled": "Enabled",
1012
1126
"output"))
1013
1127
1014
1128
1015
class Unique(object):
1129
class Unique:
1016
1130
"""Class for objects which exist only to be unique objects, since
1017
1131
unittest.mock.sentinel only exists in Python 3.3"""
1018
1132
1302
1416
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1303
1417
1304
1418
def MockDBusPython_func(self, func):
1305
class mock_dbus_python(object):
1419
class mock_dbus_python:
1306
1420
"""mock dbus-python module"""
1307
class exceptions(object):
1421
class exceptions:
1308
1422
"""Pseudo-namespace"""
1309
1423
class DBusException(Exception):
1310
1424
pass
1311
class SystemBus(object):
1425
class SystemBus:
1312
1426
@staticmethod
1313
1427
def get_object(busname, objectpath):
1314
1428
DBusObject = collections.namedtuple(
1315
"DBusObject", ("methodname",))
1429
"DBusObject", ("methodname", "Set"))
1316
1430
def method(*args, **kwargs):
1317
1431
self.assertEqual({"dbus_interface":
1318
1432
"interface"},
1319
1433
kwargs)
1320
1434
return func(*args)
1321
return DBusObject(methodname=method)
1322
class Boolean(object):
1435
def set_property(interface, key, value,
1436
dbus_interface=None):
1437
self.assertEqual(
1438
"org.freedesktop.DBus.Properties",
1439
dbus_interface)
1440
self.assertEqual("Secret", key)
1441
return func(interface, key, value,
1442
dbus_interface=dbus_interface)
1443
return DBusObject(methodname=method,
1444
Set=set_property)
1445
class Boolean:
1323
1446
def __init__(self, value):
1324
1447
self.value = bool(value)
1325
1448
def __bool__(self):
1330
1453
pass
1331
1454
class Dictionary(dict):
1332
1455
pass
1456
class ByteArray(bytes):
1457
pass
1333
1458
return mock_dbus_python
1334
1459
1335
1460
def call_method(self, bus, methodname, busname, objectpath,
1507
1632
finally:
1508
1633
dbus_logger.removeFilter(counting_handler)
1509
1634
1510
self.assertNotIsInstance(e, dbus.ConnectFailed)
1635
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1511
1636
1512
1637
# Make sure the dbus logger was suppressed
1513
1638
self.assertEqual(0, counting_handler.count)
1514
1639
1640
def test_Set_Secret_sends_bytearray(self):
1641
ret = [None]
1642
def func(*args, **kwargs):
1643
ret[0] = (args, kwargs)
1644
mock_dbus_python = self.MockDBusPython_func(func)
1645
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1646
bus.set_client_property("objectpath", "Secret", "value")
1647
expected_call = (("se.recompile.Mandos.Client", "Secret",
1648
mock_dbus_python.ByteArray(b"value")),
1649
{"dbus_interface":
1650
"org.freedesktop.DBus.Properties"})
1651
self.assertEqual(expected_call, ret[0])
1652
if sys.version_info.major == 2:
1653
self.assertIsInstance(ret[0][0][-1],
1654
mock_dbus_python.ByteArray)
1655
1515
1656
def test_get_object_converts_to_correct_exception(self):
1516
1657
bus = dbus_python_adapter.SystemBus(
1517
1658
self.fake_dbus_python_raises_exception_on_connect)
1519
1660
self.call_method(bus, "methodname", "busname",
1520
1661
"objectpath", "interface")
1521
1662
1522
class fake_dbus_python_raises_exception_on_connect(object):
1663
class fake_dbus_python_raises_exception_on_connect:
1523
1664
"""fake dbus-python module"""
1524
class exceptions(object):
1665
class exceptions:
1525
1666
"""Pseudo-namespace"""
1526
1667
class DBusException(Exception):
1527
1668
pass
1535
1676
1536
1677
1537
1678
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1538
class mock_dbus_python(object):
1679
class mock_dbus_python:
1539
1680
"""mock dbus-python modules"""
1540
class SystemBus(object):
1681
class SystemBus:
1541
1682
@staticmethod
1542
1683
def get_object(busname, objectpath):
1543
1684
return Unique()
1589
1730
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1590
1731
1591
1732
def Stub_pydbus_func(self, func):
1592
class stub_pydbus(object):
1733
class stub_pydbus:
1593
1734
"""stub pydbus module"""
1594
class SystemBus(object):
1735
class SystemBus:
1595
1736
@staticmethod
1596
1737
def get(busname, objectpath):
1597
1738
DBusObject = collections.namedtuple(
1634
1775
self.call_method(bus, "methodname", "busname",
1635
1776
"objectpath", "interface")
1636
1777
1637
self.assertNotIsInstance(e, dbus.ConnectFailed)
1778
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1638
1779
1639
1780
def test_get_converts_to_correct_exception(self):
1640
1781
bus = pydbus_adapter.SystemBus(
1643
1784
self.call_method(bus, "methodname", "busname",
1644
1785
"objectpath", "interface")
1645
1786
1646
class fake_pydbus_raises_exception_on_connect(object):
1787
class fake_pydbus_raises_exception_on_connect:
1647
1788
"""fake dbus-python module"""
1648
1789
@classmethod
1649
1790
def SystemBus(cls):
1653
1794
return Bus(get=get)
1654
1795
1655
1796
def test_set_property_uses_setattr(self):
1656
class Object(object):
1797
class Object:
1657
1798
pass
1658
1799
obj = Object()
1659
class pydbus_spy(object):
1660
class SystemBus(object):
1800
class pydbus_spy:
1801
class SystemBus:
1661
1802
@staticmethod
1662
1803
def get(busname, objectpath):
1663
1804
return {"interface": obj}
1670
1811
def test_get_suppresses_xml_deprecation_warning(self):
1671
1812
if sys.version_info.major >= 3:
1672
1813
return
1673
class stub_pydbus_get(object):
1674
class SystemBus(object):
1814
class stub_pydbus_get:
1815
class SystemBus:
1675
1816
@staticmethod
1676
1817
def get(busname, objectpath):
1677
1818
warnings.warn_explicit(
1685
1826
1686
1827
1687
1828
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1688
class stub_pydbus(object):
1829
class stub_pydbus:
1689
1830
"""stub pydbus module"""
1690
class SystemBus(object):
1831
class SystemBus:
1691
1832
@staticmethod
1692
1833
def get(busname, objectpath):
1693
1834
return Unique()
1735
1876
self.assertIs(obj1, obj1b)
1736
1877
1737
1878
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
1738
2061
class Test_commands_from_options(unittest.TestCase):
1739
2062
1740
2063
def setUp(self):
1745
2068
self.assert_command_from_args(["--is-enabled", "client"],
1746
2069
command.IsEnabled)
1747
2070
1748
def assert_command_from_args(self, args, command_cls,
1749
**cmd_attrs):
2071
def assert_command_from_args(self, args, command_cls, length=1,
2072
clients=None, **cmd_attrs):
1750
2073
"""Assert that parsing ARGS should result in an instance of
1751
2074
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1752
2075
options = self.parser.parse_args(args)
1753
2076
check_option_syntax(self.parser, options)
1754
2077
commands = commands_from_options(options)
1755
self.assertEqual(1, len(commands))
1756
command = commands[0]
1757
self.assertIsInstance(command, command_cls)
2078
self.assertEqual(length, len(commands))
2079
for command in commands:
2080
if isinstance(command, command_cls):
2081
break
2082
else:
2083
self.assertIsInstance(command, command_cls)
2084
if clients is not None:
2085
self.assertEqual(clients, options.client)
1758
2086
for key, value in cmd_attrs.items():
1759
2087
self.assertEqual(value, getattr(command, key))
1760
2088
2089
def assert_commands_from_args(self, args, commands, clients=None):
2090
for cmd in commands:
2091
self.assert_command_from_args(args, cmd,
2092
length=len(commands),
2093
clients=clients)
2094
1761
2095
def test_is_enabled_short(self):
1762
2096
self.assert_command_from_args(["-V", "client"],
1763
2097
command.IsEnabled)
1954
2288
verbose=True)
1955
2289
1956
2290
2291
def test_manual_page_example_1(self):
2292
self.assert_command_from_args("",
2293
command.PrintTable,
2294
clients=[],
2295
verbose=False)
2296
2297
def test_manual_page_example_2(self):
2298
self.assert_command_from_args(
2299
"--verbose foo1.example.org foo2.example.org".split(),
2300
command.PrintTable, clients=["foo1.example.org",
2301
"foo2.example.org"],
2302
verbose=True)
2303
2304
def test_manual_page_example_3(self):
2305
self.assert_command_from_args("--enable --all".split(),
2306
command.Enable,
2307
clients=[])
2308
2309
def test_manual_page_example_4(self):
2310
self.assert_commands_from_args(
2311
("--timeout=PT5M --interval=PT1M foo1.example.org"
2312
" foo2.example.org").split(),
2313
[command.SetTimeout, command.SetInterval],
2314
clients=["foo1.example.org", "foo2.example.org"])
2315
2316
def test_manual_page_example_5(self):
2317
self.assert_command_from_args("--approve --all".split(),
2318
command.Approve,
2319
clients=[])
2320
2321
1957
2322
class TestCommand(unittest.TestCase):
1958
2323
"""Abstract class for tests of command classes"""
1959
2324
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0