/mandos/trunk : revision 1152

1

#!/usr/bin/python3 -bbI

1

#!/usr/bin/python

2

# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

46

import tempfile

47

import contextlib

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

64

49

try:

65

import dbussy

66

import ravel

50

import pydbus

51

import gi

52

dbus_python = None

67

53

except ImportError:

68

try:

69

import pydbus

70

import gi

71

except ImportError:

72

import dbus as dbus_python

73

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

74

62

75

63

# Show warnings by default

76

64

if not sys.warnoptions:

84

72

logging.captureWarnings(True) # Show warnings via the logging system

85

73

86

74

if sys.version_info.major == 2:

75

str = unicode

87

76

import StringIO

88

77

io.StringIO = StringIO.StringIO

89

78

90

79

locale.setlocale(locale.LC_ALL, "")

91

80

92

version = "1.8.9"

81

version = "1.8.6"

93

82

94

83

95

84

def main():

104

93

if options.debug:

105

94

log.setLevel(logging.DEBUG)

106

95

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

109

elif pydbus is not None:

96

if pydbus is not None:

110

97

bus = pydbus_adapter.CachingBus(pydbus)

111

98

else:

112

99

bus = dbus_python_adapter.CachingBus(dbus_python)

480

467

parser.error("--remove can only be combined with --deny")

481

468

482

469

483

class dbus:

470

class dbus(object):

484

471

485

class SystemBus:

472

class SystemBus(object):

486

473

487

474

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

488

475

def get_managed_objects(self, busname, objectpath):

497

484

self.properties_iface, interface, key,

498

485

value)

499

486

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

504

487

505

488

class MandosBus(SystemBus):

506

489

busname_domain = "se.recompile"

538

521

pass

539

522

540

523

541

class dbus_python_adapter:

524

class dbus_python_adapter(object):

542

525

543

526

class SystemBus(dbus.MandosBus):

544

527

"""Use dbus-python"""

598

581

self.client_interface, key,

599

582

value)

600

583

601

class SilenceLogger:

584

class SilenceLogger(object):

602

585

"Simple context manager to silence a particular logger"

603

586

def __init__(self, loggername):

604

587

self.logger = logging.getLogger(loggername)

633

616

return new_object

634

617

635

618

636

class pydbus_adapter:

619

class pydbus_adapter(object):

637

620

class SystemBus(dbus.MandosBus):

638

621

def __init__(self, module=pydbus):

639

622

self.pydbus = module

696

679

return new_object

697

680

698

681

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

682

def commands_from_options(options):

788

683

789

684

commands = list(options.commands)

817

712

return commands

818

713

819

714

820

class command:

715

class command(object):

821

716

"""A namespace for command classes"""

822

717

823

class Base:

718

class Base(object):

824

719

"""Abstract base class for commands"""

825

720

def run(self, clients, bus=None):

826

721

"""Normal commands should implement run_on_one_client(),

889

784

keywords = self.all_keywords

890

785

print(self.TableOfClients(clients.values(), keywords))

891

786

892

class TableOfClients:

787

class TableOfClients(object):

893

788

tableheaders = {

894

789

"Name": "Name",

895

790

"Enabled": "Enabled",

1126

1021

"output"))

1127

1022

1128

1023

1129

class Unique:

1024

class Unique(object):

1130

1025

"""Class for objects which exist only to be unique objects, since

1131

1026

unittest.mock.sentinel only exists in Python 3.3"""

1132

1027

1416

1311

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1417

1312

1418

1313

def MockDBusPython_func(self, func):

1419

class mock_dbus_python:

1314

class mock_dbus_python(object):

1420

1315

"""mock dbus-python module"""

1421

class exceptions:

1316

class exceptions(object):

1422

1317

"""Pseudo-namespace"""

1423

1318

class DBusException(Exception):

1424

1319

pass

1425

class SystemBus:

1320

class SystemBus(object):

1426

1321

@staticmethod

1427

1322

def get_object(busname, objectpath):

1428

1323

DBusObject = collections.namedtuple(

1442

1337

dbus_interface=dbus_interface)

1443

1338

return DBusObject(methodname=method,

1444

1339

Set=set_property)

1445

class Boolean:

1340

class Boolean(object):

1446

1341

def __init__(self, value):

1447

1342

self.value = bool(value)

1448

1343

def __bool__(self):

1632

1527

finally:

1633

1528

dbus_logger.removeFilter(counting_handler)

1634

1529

1635

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

1530

self.assertNotIsInstance(e, dbus.ConnectFailed)

1636

1531

1637

1532

# Make sure the dbus logger was suppressed

1638

1533

self.assertEqual(0, counting_handler.count)

1660

1555

self.call_method(bus, "methodname", "busname",

1661

1556

"objectpath", "interface")

1662

1557

1663

class fake_dbus_python_raises_exception_on_connect:

1558

class fake_dbus_python_raises_exception_on_connect(object):

1664

1559

"""fake dbus-python module"""

1665

class exceptions:

1560

class exceptions(object):

1666

1561

"""Pseudo-namespace"""

1667

1562

class DBusException(Exception):

1668

1563

pass

1676

1571

1677

1572

1678

1573

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1679

class mock_dbus_python:

1574

class mock_dbus_python(object):

1680

1575

"""mock dbus-python modules"""

1681

class SystemBus:

1576

class SystemBus(object):

1682

1577

@staticmethod

1683

1578

def get_object(busname, objectpath):

1684

1579

return Unique()

1730

1625

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1731

1626

1732

1627

def Stub_pydbus_func(self, func):

1733

class stub_pydbus:

1628

class stub_pydbus(object):

1734

1629

"""stub pydbus module"""

1735

class SystemBus:

1630

class SystemBus(object):

1736

1631

@staticmethod

1737

1632

def get(busname, objectpath):

1738

1633

DBusObject = collections.namedtuple(

1775

1670

self.call_method(bus, "methodname", "busname",

1776

1671

"objectpath", "interface")

1777

1672

1778

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1779

1674

1780

1675

def test_get_converts_to_correct_exception(self):

1781

1676

bus = pydbus_adapter.SystemBus(

1784

1679

self.call_method(bus, "methodname", "busname",

1785

1680

"objectpath", "interface")

1786

1681

1787

class fake_pydbus_raises_exception_on_connect:

1682

class fake_pydbus_raises_exception_on_connect(object):

1788

1683

"""fake dbus-python module"""

1789

1684

@classmethod

1790

1685

def SystemBus(cls):

1794

1689

return Bus(get=get)

1795

1690

1796

1691

def test_set_property_uses_setattr(self):

1797

class Object:

1692

class Object(object):

1798

1693

pass

1799

1694

obj = Object()

1800

class pydbus_spy:

1801

class SystemBus:

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1802

1697

@staticmethod

1803

1698

def get(busname, objectpath):

1804

1699

return {"interface": obj}

1811

1706

def test_get_suppresses_xml_deprecation_warning(self):

1812

1707

if sys.version_info.major >= 3:

1813

1708

return

1814

class stub_pydbus_get:

1815

class SystemBus:

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1816

1711

@staticmethod

1817

1712

def get(busname, objectpath):

1818

1713

warnings.warn_explicit(

1826

1721

1827

1722

1828

1723

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1829

class stub_pydbus:

1724

class stub_pydbus(object):

1830

1725

"""stub pydbus module"""

1831

class SystemBus:

1726

class SystemBus(object):

1832

1727

@staticmethod

1833

1728

def get(busname, objectpath):

1834

1729

return Unique()

1876

1771

self.assertIs(obj1, obj1b)

1877

1772

1878

1773

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

1774

class Test_commands_from_options(unittest.TestCase):

2062

1775

2063

1776

def setUp(self):