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
- browse files at revision 1243
- compare with another revision
- download diff
- download tarball
- view history from revision 1243
- Committer: Teddy Hogeborn
- Date: 2021-03-21 20:46:40 UTC
- Revision ID: teddy@recompile.se-20210321204640-lpsyen8jr9lw1jma
Some cleanup of GnuTLS interface
Rename opaque internal GnuTLS structures named *_int to also start
with underscore (_), as is the custom in Python programs.
Decode byte strings from UTF-8 where needed. (Fixing, among other
things, all "DEBUG: GnuTLS" lines having a "b'" prefix in Python 3.)
Simplify calling C functions by:
1. Using the "_as_parameter_" attribute to store the ctypes object.
2. Creating and using helper classes to automatically create pointers
or cast typed pointers to pointers to void.
3. Providing the "from_param()" method on relevant classes.
Remove "restype" attribute on C functions where "errcheck" attribute
is already set.
* mandos (gnutls.session_int): Rename to start with "_".
(gnutls.openpgp_crt_int): - '' -
(gnutls.Error.__init__): Decode byte string from gnutls.strerror().
(gnutls.PointerTo): New helper class.
(gnutls.CastToVoidPointer): - '' -
(gnutls.With_from_param): - '' -
(gnutls.Credentials): Inherit from "With_from_param" and store the
ctypes object in the "_as_parameter_" attribute instead of
"_c_object".
(gnutls._error_code): Use "gnutls.E_SUCCESS" instead of the unadorned
numerical constant "0".
(gnutls._retry_on_error): - '' -
(gnutls.priority_set_direct.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(gnutls.init.argtypes): Use "PointerTo(ClientSession)" instead of
"ctypes.POINTER(session_t)", and change all callers to match.
(gnutls.set_default_priority.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(gnutls.record_send.argtypes): - '' -
(gnutls.certificate_allocate_credentials.argtypes): Use
"PointerTo(Credentials)" instead of
"ctypes.POINTER(certificate_credentials_t)", and change all callers
to match.
(gnutls.certificate_free_credentials.argtypes): Use "Credentials"
instead of "certificate_credentials_t", and change all callers to
match.
(gnutls.handshake_set_private_extensions.argtypes): Use
"ClientSession" instead of "session_t", and change all callers to
match.
(gnutls.credentials_set.argtypes): Use
"CastToVoidPointer(Credentials)" instead of "ctypes.c_void_p", and
change all callers to match.
(gnutls.certificate_type_get.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(gnutls.certificate_get_peers.argtypes): - '' -
(gnutls.deinit.argtypes): - '' -
(gnutls.handshake.argtypes): - '' -
(gnutls.handshake.restype): Change from "_error_code" to
"ctypes.c_int".
(gnutls.transport_set_ptr.argtypes): Use "ClientSession" instead of
"session_t", and change all callers to match.
(gnutls.bye.argtypes): - '' -
(gnutls.bye.restype): Change from "_error_code" to "ctypes.c_int".
(gnutls.certificate_type_get2.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(ClientHandler.handle): Decode "key_id" bytes to string before
logging it in the debug log.
(main.debug_gnutls): Decode GnuTLS log message from bytes to string
before logging it in the debug log.
Rename opaque internal GnuTLS structures named *_int to also start
with underscore (_), as is the custom in Python programs.
Decode byte strings from UTF-8 where needed. (Fixing, among other
things, all "DEBUG: GnuTLS" lines having a "b'" prefix in Python 3.)
Simplify calling C functions by:
1. Using the "_as_parameter_" attribute to store the ctypes object.
2. Creating and using helper classes to automatically create pointers
or cast typed pointers to pointers to void.
3. Providing the "from_param()" method on relevant classes.
Remove "restype" attribute on C functions where "errcheck" attribute
is already set.
* mandos (gnutls.session_int): Rename to start with "_".
(gnutls.openpgp_crt_int): - '' -
(gnutls.Error.__init__): Decode byte string from gnutls.strerror().
(gnutls.PointerTo): New helper class.
(gnutls.CastToVoidPointer): - '' -
(gnutls.With_from_param): - '' -
(gnutls.Credentials): Inherit from "With_from_param" and store the
ctypes object in the "_as_parameter_" attribute instead of
"_c_object".
(gnutls._error_code): Use "gnutls.E_SUCCESS" instead of the unadorned
numerical constant "0".
(gnutls._retry_on_error): - '' -
(gnutls.priority_set_direct.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(gnutls.init.argtypes): Use "PointerTo(ClientSession)" instead of
"ctypes.POINTER(session_t)", and change all callers to match.
(gnutls.set_default_priority.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(gnutls.record_send.argtypes): - '' -
(gnutls.certificate_allocate_credentials.argtypes): Use
"PointerTo(Credentials)" instead of
"ctypes.POINTER(certificate_credentials_t)", and change all callers
to match.
(gnutls.certificate_free_credentials.argtypes): Use "Credentials"
instead of "certificate_credentials_t", and change all callers to
match.
(gnutls.handshake_set_private_extensions.argtypes): Use
"ClientSession" instead of "session_t", and change all callers to
match.
(gnutls.credentials_set.argtypes): Use
"CastToVoidPointer(Credentials)" instead of "ctypes.c_void_p", and
change all callers to match.
(gnutls.certificate_type_get.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(gnutls.certificate_get_peers.argtypes): - '' -
(gnutls.deinit.argtypes): - '' -
(gnutls.handshake.argtypes): - '' -
(gnutls.handshake.restype): Change from "_error_code" to
"ctypes.c_int".
(gnutls.transport_set_ptr.argtypes): Use "ClientSession" instead of
"session_t", and change all callers to match.
(gnutls.bye.argtypes): - '' -
(gnutls.bye.restype): Change from "_error_code" to "ctypes.c_int".
(gnutls.certificate_type_get2.argtypes): Use "ClientSession" instead
of "session_t", and change all callers to match.
(ClientHandler.handle): Decode "key_id" bytes to string before
logging it in the debug log.
(main.debug_gnutls): Decode GnuTLS log message from bytes to string
before logging it in the debug log.
- 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*")))); -*-
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
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
#
4
# Mandos Control - Control or query the Mandos server
5
#
6
# Copyright © 2008-2020 Teddy Hogeborn
7
# Copyright © 2008-2020 Björn Påhlsson
8
8
#
9
9
# This file is part of Mandos.
10
10
#
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.14"
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
# list of one element with the return 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
# A D-Bus Variant value is represented as the Python type
742
# Tuple[dbussy.DBUS.Signature, Any]
743
if isinstance(value, tuple):
744
if (len(value) == 2
745
and isinstance(value[0],
746
self.dbussy.DBUS.Signature)):
747
return self.type_filter(value[1])
748
elif isinstance(value, self.dbussy.DBUS.ObjectPath):
749
return str(value)
750
# Also recurse into dictionaries
751
elif isinstance(value, dict):
752
return {self.type_filter(key):
753
self.type_filter(subval)
754
for key, subval in value.items()}
755
return value
756
757
def set_property(self, busname, objectpath, interface, key,
758
value):
759
proxy_object = self.get_object(busname, objectpath)
760
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
761
objectpath, self.properties_iface, interface,
762
key, value)
763
if key == "Secret":
764
# DBussy wants a Byte Array to be a sequence of
765
# values, not a byte string
766
value = tuple(value)
767
setattr(proxy_object.get_interface(interface), key, value)
768
769
class MandosBus(SystemBus, dbus.MandosBus):
770
pass
771
772
class CachingBus(MandosBus):
773
"""A caching layer for dbussy_adapter.MandosBus"""
774
def __init__(self, *args, **kwargs):
775
self.object_cache = {}
776
super(dbussy_adapter.CachingBus, self).__init__(*args,
777
**kwargs)
778
def get_object(self, busname, objectpath):
779
try:
780
return self.object_cache[(busname, objectpath)]
781
except KeyError:
782
new_object = super(
783
dbussy_adapter.CachingBus,
784
self).get_object(busname, objectpath)
785
self.object_cache[(busname, objectpath)] = new_object
786
return new_object
787
788
673
789
def commands_from_options(options):
674
790
675
791
commands = list(options.commands)
703
819
return commands
704
820
705
821
706
class command(object):
822
class command:
707
823
"""A namespace for command classes"""
708
824
709
class Base(object):
825
class Base:
710
826
"""Abstract base class for commands"""
711
827
def run(self, clients, bus=None):
712
828
"""Normal commands should implement run_on_one_client(),
775
891
keywords = self.all_keywords
776
892
print(self.TableOfClients(clients.values(), keywords))
777
893
778
class TableOfClients(object):
894
class TableOfClients:
779
895
tableheaders = {
780
896
"Name": "Name",
781
897
"Enabled": "Enabled",
1012
1128
"output"))
1013
1129
1014
1130
1015
class Unique(object):
1131
class Unique:
1016
1132
"""Class for objects which exist only to be unique objects, since
1017
1133
unittest.mock.sentinel only exists in Python 3.3"""
1018
1134
1302
1418
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1303
1419
1304
1420
def MockDBusPython_func(self, func):
1305
class mock_dbus_python(object):
1421
class mock_dbus_python:
1306
1422
"""mock dbus-python module"""
1307
class exceptions(object):
1423
class exceptions:
1308
1424
"""Pseudo-namespace"""
1309
1425
class DBusException(Exception):
1310
1426
pass
1311
class SystemBus(object):
1427
class SystemBus:
1312
1428
@staticmethod
1313
1429
def get_object(busname, objectpath):
1314
1430
DBusObject = collections.namedtuple(
1315
"DBusObject", ("methodname",))
1431
"DBusObject", ("methodname", "Set"))
1316
1432
def method(*args, **kwargs):
1317
1433
self.assertEqual({"dbus_interface":
1318
1434
"interface"},
1319
1435
kwargs)
1320
1436
return func(*args)
1321
return DBusObject(methodname=method)
1322
class Boolean(object):
1437
def set_property(interface, key, value,
1438
dbus_interface=None):
1439
self.assertEqual(
1440
"org.freedesktop.DBus.Properties",
1441
dbus_interface)
1442
self.assertEqual("Secret", key)
1443
return func(interface, key, value,
1444
dbus_interface=dbus_interface)
1445
return DBusObject(methodname=method,
1446
Set=set_property)
1447
class Boolean:
1323
1448
def __init__(self, value):
1324
1449
self.value = bool(value)
1325
1450
def __bool__(self):
1330
1455
pass
1331
1456
class Dictionary(dict):
1332
1457
pass
1458
class ByteArray(bytes):
1459
pass
1333
1460
return mock_dbus_python
1334
1461
1335
1462
def call_method(self, bus, methodname, busname, objectpath,
1507
1634
finally:
1508
1635
dbus_logger.removeFilter(counting_handler)
1509
1636
1510
self.assertNotIsInstance(e, dbus.ConnectFailed)
1637
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1511
1638
1512
1639
# Make sure the dbus logger was suppressed
1513
1640
self.assertEqual(0, counting_handler.count)
1514
1641
1642
def test_Set_Secret_sends_bytearray(self):
1643
ret = [None]
1644
def func(*args, **kwargs):
1645
ret[0] = (args, kwargs)
1646
mock_dbus_python = self.MockDBusPython_func(func)
1647
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1648
bus.set_client_property("objectpath", "Secret", "value")
1649
expected_call = (("se.recompile.Mandos.Client", "Secret",
1650
mock_dbus_python.ByteArray(b"value")),
1651
{"dbus_interface":
1652
"org.freedesktop.DBus.Properties"})
1653
self.assertEqual(expected_call, ret[0])
1654
if sys.version_info.major == 2:
1655
self.assertIsInstance(ret[0][0][-1],
1656
mock_dbus_python.ByteArray)
1657
1515
1658
def test_get_object_converts_to_correct_exception(self):
1516
1659
bus = dbus_python_adapter.SystemBus(
1517
1660
self.fake_dbus_python_raises_exception_on_connect)
1519
1662
self.call_method(bus, "methodname", "busname",
1520
1663
"objectpath", "interface")
1521
1664
1522
class fake_dbus_python_raises_exception_on_connect(object):
1665
class fake_dbus_python_raises_exception_on_connect:
1523
1666
"""fake dbus-python module"""
1524
class exceptions(object):
1667
class exceptions:
1525
1668
"""Pseudo-namespace"""
1526
1669
class DBusException(Exception):
1527
1670
pass
1535
1678
1536
1679
1537
1680
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1538
class mock_dbus_python(object):
1681
class mock_dbus_python:
1539
1682
"""mock dbus-python modules"""
1540
class SystemBus(object):
1683
class SystemBus:
1541
1684
@staticmethod
1542
1685
def get_object(busname, objectpath):
1543
1686
return Unique()
1589
1732
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1590
1733
1591
1734
def Stub_pydbus_func(self, func):
1592
class stub_pydbus(object):
1735
class stub_pydbus:
1593
1736
"""stub pydbus module"""
1594
class SystemBus(object):
1737
class SystemBus:
1595
1738
@staticmethod
1596
1739
def get(busname, objectpath):
1597
1740
DBusObject = collections.namedtuple(
1634
1777
self.call_method(bus, "methodname", "busname",
1635
1778
"objectpath", "interface")
1636
1779
1637
self.assertNotIsInstance(e, dbus.ConnectFailed)
1780
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1638
1781
1639
1782
def test_get_converts_to_correct_exception(self):
1640
1783
bus = pydbus_adapter.SystemBus(
1643
1786
self.call_method(bus, "methodname", "busname",
1644
1787
"objectpath", "interface")
1645
1788
1646
class fake_pydbus_raises_exception_on_connect(object):
1789
class fake_pydbus_raises_exception_on_connect:
1647
1790
"""fake dbus-python module"""
1648
1791
@classmethod
1649
1792
def SystemBus(cls):
1653
1796
return Bus(get=get)
1654
1797
1655
1798
def test_set_property_uses_setattr(self):
1656
class Object(object):
1799
class Object:
1657
1800
pass
1658
1801
obj = Object()
1659
class pydbus_spy(object):
1660
class SystemBus(object):
1802
class pydbus_spy:
1803
class SystemBus:
1661
1804
@staticmethod
1662
1805
def get(busname, objectpath):
1663
1806
return {"interface": obj}
1670
1813
def test_get_suppresses_xml_deprecation_warning(self):
1671
1814
if sys.version_info.major >= 3:
1672
1815
return
1673
class stub_pydbus_get(object):
1674
class SystemBus(object):
1816
class stub_pydbus_get:
1817
class SystemBus:
1675
1818
@staticmethod
1676
1819
def get(busname, objectpath):
1677
1820
warnings.warn_explicit(
1685
1828
1686
1829
1687
1830
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1688
class stub_pydbus(object):
1831
class stub_pydbus:
1689
1832
"""stub pydbus module"""
1690
class SystemBus(object):
1833
class SystemBus:
1691
1834
@staticmethod
1692
1835
def get(busname, objectpath):
1693
1836
return Unique()
1735
1878
self.assertIs(obj1, obj1b)
1736
1879
1737
1880
1881
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1882
1883
class dummy_dbussy:
1884
class DBUS:
1885
class ObjectPath(str):
1886
pass
1887
class DBusError(Exception):
1888
pass
1889
1890
def fake_ravel_func(self, func):
1891
class fake_ravel:
1892
@staticmethod
1893
def system_bus():
1894
class DBusInterfaceProxy:
1895
@staticmethod
1896
def methodname(*args):
1897
return [func(*args)]
1898
class DBusObject:
1899
@staticmethod
1900
def get_interface(interface):
1901
if interface == "interface":
1902
return DBusInterfaceProxy()
1903
return {"busname": {"objectpath": DBusObject()}}
1904
return fake_ravel
1905
1906
def call_method(self, bus, methodname, busname, objectpath,
1907
interface, *args):
1908
with self.assertLogs(log, logging.DEBUG):
1909
return bus.call_method(methodname, busname, objectpath,
1910
interface, *args)
1911
1912
def test_call_method_returns(self):
1913
expected_method_return = Unique()
1914
method_args = (Unique(), Unique())
1915
def func(*args):
1916
self.assertEqual(len(method_args), len(args))
1917
for marg, arg in zip(method_args, args):
1918
self.assertIs(marg, arg)
1919
return expected_method_return
1920
fake_ravel = self.fake_ravel_func(func)
1921
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1922
ret = self.call_method(bus, "methodname", "busname",
1923
"objectpath", "interface",
1924
*method_args)
1925
self.assertIs(ret, expected_method_return)
1926
1927
def test_call_method_filters_objectpath(self):
1928
def func():
1929
return method_return
1930
fake_ravel = self.fake_ravel_func(func)
1931
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1932
method_return = (self.dummy_dbussy.DBUS
1933
.ObjectPath("objectpath"))
1934
ret = self.call_method(bus, "methodname", "busname",
1935
"objectpath", "interface")
1936
self.assertEqual("objectpath", ret)
1937
self.assertNotIsInstance(ret,
1938
self.dummy_dbussy.DBUS.ObjectPath)
1939
1940
def test_call_method_filters_objectpaths_in_dict(self):
1941
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1942
def func():
1943
return method_return
1944
fake_ravel = self.fake_ravel_func(func)
1945
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1946
method_return = {
1947
ObjectPath("objectpath_key_1"):
1948
ObjectPath("objectpath_value_1"),
1949
ObjectPath("objectpath_key_2"):
1950
ObjectPath("objectpath_value_2"),
1951
}
1952
ret = self.call_method(bus, "methodname", "busname",
1953
"objectpath", "interface")
1954
expected_method_return = {str(key): str(value)
1955
for key, value in
1956
method_return.items()}
1957
for key, value in ret.items():
1958
self.assertNotIsInstance(key, ObjectPath)
1959
self.assertNotIsInstance(value, ObjectPath)
1960
self.assertEqual(expected_method_return, ret)
1961
self.assertIsInstance(ret, dict)
1962
1963
def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1964
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1965
def func():
1966
return method_return
1967
fake_ravel = self.fake_ravel_func(func)
1968
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1969
method_return = {
1970
ObjectPath("key1"): {
1971
ObjectPath("key11"): ObjectPath("value11"),
1972
ObjectPath("key12"): ObjectPath("value12"),
1973
},
1974
ObjectPath("key2"): {
1975
ObjectPath("key21"): ObjectPath("value21"),
1976
ObjectPath("key22"): ObjectPath("value22"),
1977
},
1978
}
1979
ret = self.call_method(bus, "methodname", "busname",
1980
"objectpath", "interface")
1981
expected_method_return = {
1982
"key1": {"key11": "value11",
1983
"key12": "value12"},
1984
"key2": {"key21": "value21",
1985
"key22": "value22"},
1986
}
1987
self.assertEqual(expected_method_return, ret)
1988
for key, value in ret.items():
1989
self.assertIsInstance(value, dict)
1990
self.assertEqual(expected_method_return[key], value)
1991
self.assertNotIsInstance(key, ObjectPath)
1992
for inner_key, inner_value in value.items():
1993
self.assertIsInstance(value, dict)
1994
self.assertEqual(
1995
expected_method_return[key][inner_key],
1996
inner_value)
1997
self.assertNotIsInstance(key, ObjectPath)
1998
1999
def test_call_method_filters_objectpaths_in_dict_three_deep(self):
2000
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
2001
def func():
2002
return method_return
2003
fake_ravel = self.fake_ravel_func(func)
2004
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2005
method_return = {
2006
ObjectPath("key1"): {
2007
ObjectPath("key2"): {
2008
ObjectPath("key3"): ObjectPath("value"),
2009
},
2010
},
2011
}
2012
ret = self.call_method(bus, "methodname", "busname",
2013
"objectpath", "interface")
2014
expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2015
self.assertEqual(expected_method_return, ret)
2016
self.assertIsInstance(ret, dict)
2017
self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2018
self.assertIsInstance(ret["key1"], dict)
2019
self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2020
ObjectPath)
2021
self.assertIsInstance(ret["key1"]["key2"], dict)
2022
self.assertNotIsInstance(
2023
next(iter(ret["key1"]["key2"].keys())),
2024
ObjectPath)
2025
self.assertEqual("value", ret["key1"]["key2"]["key3"])
2026
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2027
self.dummy_dbussy.DBUS.ObjectPath)
2028
2029
def test_call_method_handles_exception(self):
2030
def func():
2031
raise self.dummy_dbussy.DBusError()
2032
2033
fake_ravel = self.fake_ravel_func(func)
2034
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2035
2036
with self.assertRaises(dbus.Error) as e:
2037
self.call_method(bus, "methodname", "busname",
2038
"objectpath", "interface")
2039
2040
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2041
2042
def test_get_object_converts_to_correct_exception(self):
2043
class fake_ravel_raises_exception_on_connect:
2044
@staticmethod
2045
def system_bus():
2046
class Bus:
2047
@staticmethod
2048
def __getitem__(key):
2049
if key == "objectpath":
2050
raise self.dummy_dbussy.DBusError()
2051
raise Exception(key)
2052
return {"busname": Bus()}
2053
def func():
2054
raise self.dummy_dbussy.DBusError()
2055
bus = dbussy_adapter.SystemBus(
2056
self.dummy_dbussy,
2057
fake_ravel_raises_exception_on_connect)
2058
with self.assertRaises(dbus.ConnectFailed):
2059
self.call_method(bus, "methodname", "busname",
2060
"objectpath", "interface")
2061
2062
1738
2063
class Test_commands_from_options(unittest.TestCase):
1739
2064
1740
2065
def setUp(self):
1745
2070
self.assert_command_from_args(["--is-enabled", "client"],
1746
2071
command.IsEnabled)
1747
2072
1748
def assert_command_from_args(self, args, command_cls,
1749
**cmd_attrs):
2073
def assert_command_from_args(self, args, command_cls, length=1,
2074
clients=None, **cmd_attrs):
1750
2075
"""Assert that parsing ARGS should result in an instance of
1751
2076
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1752
2077
options = self.parser.parse_args(args)
1753
2078
check_option_syntax(self.parser, options)
1754
2079
commands = commands_from_options(options)
1755
self.assertEqual(1, len(commands))
1756
command = commands[0]
1757
self.assertIsInstance(command, command_cls)
2080
self.assertEqual(length, len(commands))
2081
for command in commands:
2082
if isinstance(command, command_cls):
2083
break
2084
else:
2085
self.assertIsInstance(command, command_cls)
2086
if clients is not None:
2087
self.assertEqual(clients, options.client)
1758
2088
for key, value in cmd_attrs.items():
1759
2089
self.assertEqual(value, getattr(command, key))
1760
2090
2091
def assert_commands_from_args(self, args, commands, clients=None):
2092
for cmd in commands:
2093
self.assert_command_from_args(args, cmd,
2094
length=len(commands),
2095
clients=clients)
2096
1761
2097
def test_is_enabled_short(self):
1762
2098
self.assert_command_from_args(["-V", "client"],
1763
2099
command.IsEnabled)
1954
2290
verbose=True)
1955
2291
1956
2292
2293
def test_manual_page_example_1(self):
2294
self.assert_command_from_args("",
2295
command.PrintTable,
2296
clients=[],
2297
verbose=False)
2298
2299
def test_manual_page_example_2(self):
2300
self.assert_command_from_args(
2301
"--verbose foo1.example.org foo2.example.org".split(),
2302
command.PrintTable, clients=["foo1.example.org",
2303
"foo2.example.org"],
2304
verbose=True)
2305
2306
def test_manual_page_example_3(self):
2307
self.assert_command_from_args("--enable --all".split(),
2308
command.Enable,
2309
clients=[])
2310
2311
def test_manual_page_example_4(self):
2312
self.assert_commands_from_args(
2313
("--timeout=PT5M --interval=PT1M foo1.example.org"
2314
" foo2.example.org").split(),
2315
[command.SetTimeout, command.SetInterval],
2316
clients=["foo1.example.org", "foo2.example.org"])
2317
2318
def test_manual_page_example_5(self):
2319
self.assert_command_from_args("--approve --all".split(),
2320
command.Approve,
2321
clients=[])
2322
2323
1957
2324
class TestCommand(unittest.TestCase):
1958
2325
"""Abstract class for tests of command classes"""
1959
2326
2072
2439
busname = "se.recompile.Mandos"
2073
2440
client_interface = "se.recompile.Mandos.Client"
2074
2441
command.Approve().run(self.bus.clients, self.bus)
2442
self.assertTrue(self.bus.clients)
2075
2443
for clientpath in self.bus.clients:
2076
2444
self.assertIn(("Approve", busname, clientpath,
2077
2445
client_interface, (True,)), self.bus.calls)
2080
2448
busname = "se.recompile.Mandos"
2081
2449
client_interface = "se.recompile.Mandos.Client"
2082
2450
command.Deny().run(self.bus.clients, self.bus)
2451
self.assertTrue(self.bus.clients)
2083
2452
for clientpath in self.bus.clients:
2084
2453
self.assertIn(("Approve", busname, clientpath,
2085
2454
client_interface, (False,)),
2086
2455
self.bus.calls)
2087
2456
2088
2457
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()
2089
2462
command.Remove().run(self.bus.clients, self.bus)
2090
for clientpath in self.bus.clients:
2091
self.assertIn(("RemoveClient", dbus_busname,
2092
dbus_server_path, dbus_server_interface,
2463
self.assertFalse(self.bus.clients)
2464
for clientpath in orig_clients:
2465
self.assertIn(("RemoveClient", busname,
2466
server_path, server_interface,
2093
2467
(clientpath,)), self.bus.calls)
2094
2468
2095
2469
expected_json = {
2297
2671
else:
2298
2672
cmd_args = [() for x in range(len(self.values_to_get))]
2299
2673
values_to_get = self.values_to_get
2674
self.assertTrue(values_to_get)
2300
2675
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2301
2676
for clientpath in self.bus.clients:
2302
2677
self.bus.clients[clientpath][self.propname] = (
2303
2678
Unique())
2304
2679
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2680
self.assertTrue(self.bus.clients)
2305
2681
for clientpath in self.bus.clients:
2306
2682
value = (self.bus.clients[clientpath]
2307
2683
[self.propname])
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0