/mandos/trunk : revision 1300

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 --test" (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

5

#

6

7

1

#!/usr/bin/python3 -bbI

2

# -*- coding: utf-8; lexical-binding: t -*-

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

26

from __future__ import (division, absolute_import, print_function,

28

27

unicode_literals)

29

28

33

32

pass

34

33

35

34

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

37

39

import locale

38

40

import datetime

39

41

import re

40

import os

41

42

import collections

42

43

import json

43

import unittest

44

import logging

45

44

import io

46

45

import tempfile

47

46

import contextlib

48

47

48

if sys.version_info.major == 2:

49

__metaclass__ = type

50

str = unicode

51

input = raw_input

52

53

54

class gi:

55

"""Dummy gi module, for the tests"""

56

class repository:

57

class GLib:

58

class Error(Exception):

59

pass

60

61

62

dbussy = None

63

ravel = None

64

dbus_python = None

65

pydbus = None

66

49

67

try:

50

import pydbus

51

import gi

52

dbus_python = None

68

import dbussy

69

import ravel

53

70

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

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

62

77

63

78

# Show warnings by default

64

79

if not sys.warnoptions:

65

80

import warnings

66

81

warnings.simplefilter("default")

67

82

68

log = logging.getLogger(sys.argv[0])

69

logging.basicConfig(level="INFO", # Show info level messages

83

log = logging.getLogger(os.path.basename(sys.argv[0]))

84

logging.basicConfig(level="INFO", # Show info level messages

70

85

format="%(message)s") # Show basic log messages

71

86

72

87

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

73

88

74

89

if sys.version_info.major == 2:

75

str = unicode

76

90

import StringIO

77

91

io.StringIO = StringIO.StringIO

78

92

79

93

locale.setlocale(locale.LC_ALL, "")

80

94

81

version = "1.8.4"

95

version = "1.8.16"

82

96

83

97

84

98

def main():

91

105

clientnames = options.client

92

106

93

107

if options.debug:

94

log.setLevel(logging.DEBUG)

108

logging.getLogger("").setLevel(logging.DEBUG)

95

109

96

if pydbus is not None:

110

if dbussy is not None and ravel is not None:

111

bus = dbussy_adapter.CachingBus(dbussy, ravel)

112

elif pydbus is not None:

97

113

bus = pydbus_adapter.CachingBus(pydbus)

98

114

else:

99

115

bus = dbus_python_adapter.CachingBus(dbus_python)

243

259

return rfc3339_duration_to_delta(interval)

244

260

except ValueError as e:

245

261

log.warning("%s - Parsing as pre-1.6.1 interval instead",

246

' '.join(e.args))

262

" ".join(e.args))

247

263

return parse_pre_1_6_1_interval(interval)

248

264

249

265

250

266

def rfc3339_duration_to_delta(duration):

251

267

"""Parse an RFC 3339 "duration" and return a datetime.timedelta

252

268

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)

269

>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

270

True

271

>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

272

True

273

>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)

274

True

275

>>> # 60 months

276

>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)

277

True

278

>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

279

True

280

>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

281

True

282

>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

283

True

284

>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

285

True

269

286

>>> # Can not be empty:

270

287

>>> rfc3339_duration_to_delta("")

271

288

Traceback (most recent call last):

378

395

379

396

380

397

def parse_pre_1_6_1_interval(interval):

381

"""Parse an interval string as documented by Mandos before 1.6.1,

398

r"""Parse an interval string as documented by Mandos before 1.6.1,

382

399

and return a datetime.timedelta

383

400

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)

401

>>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)

402

True

403

>>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)

404

True

405

>>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)

406

True

407

>>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)

408

True

409

>>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)

410

True

411

>>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)

412

True

413

>>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)

414

True

398

415

>>> # Ignore unknown characters, allow any order and repetitions

399

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

400

datetime.timedelta(2, 480, 18000)

416

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \

417

... == datetime.timedelta(2, 480, 18000)

418

True

401

419

402

420

"""

403

421

466

484

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

467

485

468

486

469

class dbus(object):

487

class dbus:

470

488

471

class SystemBus(object):

489

class SystemBus:

472

490

473

491

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

474

493

def get_managed_objects(self, busname, objectpath):

475

494

return self.call_method("GetManagedObjects", busname,

476

495

objectpath,

477

496

self.object_manager_iface)

478

497

479

498

properties_iface = "org.freedesktop.DBus.Properties"

499

480

500

def set_property(self, busname, objectpath, interface, key,

481

501

value):

482

502

self.call_method("Set", busname, objectpath,

483

503

self.properties_iface, interface, key,

484

504

value)

485

505

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

486

509

487

510

class MandosBus(SystemBus):

488

511

busname_domain = "se.recompile"

520

543

pass

521

544

522

545

523

class dbus_python_adapter(object):

546

class dbus_python_adapter:

524

547

525

548

class SystemBus(dbus.MandosBus):

526

549

"""Use dbus-python"""

580

603

self.client_interface, key,

581

604

value)

582

605

583

class SilenceLogger(object):

606

class SilenceLogger:

584

607

"Simple context manager to silence a particular logger"

608

585

609

def __init__(self, loggername):

586

610

self.logger = logging.getLogger(loggername)

587

611

597

621

def __exit__(self, exc_type, exc_val, exc_tb):

598

622

self.logger.removeFilter(self.nullfilter)

599

623

600

601

624

class CachingBus(SystemBus):

602

625

"""A caching layer for dbus_python_adapter.SystemBus"""

626

603

627

def __init__(self, *args, **kwargs):

604

628

self.object_cache = {}

605

629

super(dbus_python_adapter.CachingBus,

606

630

self).__init__(*args, **kwargs)

631

607

632

def get_object(self, busname, objectpath):

608

633

try:

609

634

return self.object_cache[(busname, objectpath)]

611

636

new_object = super(

612

637

dbus_python_adapter.CachingBus,

613

638

self).get_object(busname, objectpath)

614

self.object_cache[(busname, objectpath)] = new_object

639

self.object_cache[(busname, objectpath)] = new_object

615

640

return new_object

616

641

617

642

618

class pydbus_adapter(object):

643

class pydbus_adapter:

619

644

class SystemBus(dbus.MandosBus):

620

645

def __init__(self, module=pydbus):

621

646

self.pydbus = module

664

689

665

690

class CachingBus(SystemBus):

666

691

"""A caching layer for pydbus_adapter.SystemBus"""

692

667

693

def __init__(self, *args, **kwargs):

668

694

self.object_cache = {}

669

695

super(pydbus_adapter.CachingBus,

670

696

self).__init__(*args, **kwargs)

697

671

698

def get(self, busname, objectpath):

672

699

try:

673

700

return self.object_cache[(busname, objectpath)]

674

701

except KeyError:

675

702

new_object = (super(pydbus_adapter.CachingBus, self)

676

703

.get(busname, objectpath))

677

self.object_cache[(busname, objectpath)] = new_object

704

self.object_cache[(busname, objectpath)] = new_object

705

return new_object

706

707

708

class dbussy_adapter:

709

class SystemBus(dbus.SystemBus):

710

"""Use DBussy"""

711

712

def __init__(self, dbussy, ravel):

713

self.dbussy = dbussy

714

self.ravel = ravel

715

self.bus = ravel.system_bus()

716

717

@contextlib.contextmanager

718

def convert_exception(self, exception_class=dbus.Error):

719

try:

720

yield

721

except self.dbussy.DBusError as e:

722

# This does what "raise from" would do

723

exc = exception_class(*e.args)

724

exc.__cause__ = e

725

raise exc

726

727

def call_method(self, methodname, busname, objectpath,

728

interface, *args):

729

proxy_object = self.get_object(busname, objectpath)

730

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

731

interface, methodname,

732

", ".join(repr(a) for a in args))

733

iface = proxy_object.get_interface(interface)

734

method = getattr(iface, methodname)

735

with self.convert_exception(dbus.Error):

736

value = method(*args)

737

# DBussy returns values either as an empty list or as a

738

# list of one element with the return value

739

if value:

740

return self.type_filter(value[0])

741

742

def get_object(self, busname, objectpath):

743

log.debug("D-Bus: Connect to: (busname=%r, path=%r)",

744

busname, objectpath)

745

with self.convert_exception(dbus.ConnectFailed):

746

return self.bus[busname][objectpath]

747

748

def type_filter(self, value):

749

"""Convert the most bothersome types to Python types"""

750

# A D-Bus Variant value is represented as the Python type

751

# Tuple[dbussy.DBUS.Signature, Any]

752

if isinstance(value, tuple):

753

if (len(value) == 2

754

and isinstance(value[0],

755

self.dbussy.DBUS.Signature)):

756

return self.type_filter(value[1])

757

elif isinstance(value, self.dbussy.DBUS.ObjectPath):

758

return str(value)

759

# Also recurse into dictionaries

760

elif isinstance(value, dict):

761

return {self.type_filter(key):

762

self.type_filter(subval)

763

for key, subval in value.items()}

764

return value

765

766

def set_property(self, busname, objectpath, interface, key,

767

value):

768

proxy_object = self.get_object(busname, objectpath)

769

log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,

770

objectpath, self.properties_iface, interface,

771

key, value)

772

if key == "Secret":

773

# DBussy wants a Byte Array to be a sequence of

774

# values, not a byte string

775

value = tuple(value)

776

setattr(proxy_object.get_interface(interface), key, value)

777

778

class MandosBus(SystemBus, dbus.MandosBus):

779

pass

780

781

class CachingBus(MandosBus):

782

"""A caching layer for dbussy_adapter.MandosBus"""

783

784

def __init__(self, *args, **kwargs):

785

self.object_cache = {}

786

super(dbussy_adapter.CachingBus, self).__init__(*args,

787

**kwargs)

788

789

def get_object(self, busname, objectpath):

790

try:

791

return self.object_cache[(busname, objectpath)]

792

except KeyError:

793

new_object = super(

794

dbussy_adapter.CachingBus,

795

self).get_object(busname, objectpath)

796

self.object_cache[(busname, objectpath)] = new_object

678

797

return new_object

679

798

680

799

711

830

return commands

712

831

713

832

714

class command(object):

833

class command:

715

834

"""A namespace for command classes"""

716

835

717

class Base(object):

836

class Base:

718

837

"""Abstract base class for commands"""

838

719

839

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

720

840

"""Normal commands should implement run_on_one_client(),

721

841

but commands which want to operate on all clients at the same time can

725

845

for client, properties in clients.items():

726

846

self.run_on_one_client(client, properties)

727

847

728

729

848

class IsEnabled(Base):

730

849

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

731

850

properties = next(iter(clients.values()))

733

852

sys.exit(0)

734

853

sys.exit(1)

735

854

736

737

855

class Approve(Base):

738

856

def run_on_one_client(self, client, properties):

739

857

self.bus.call_client_method(client, "Approve", True)

740

858

741

742

859

class Deny(Base):

743

860

def run_on_one_client(self, client, properties):

744

861

self.bus.call_client_method(client, "Approve", False)

745

862

746

747

863

class Remove(Base):

748

864

def run(self, clients, bus):

749

865

for clientpath in frozenset(clients.keys()):

750

866

bus.call_server_method("RemoveClient", clientpath)

751

867

752

753

868

class Output(Base):

754

869

"""Abstract class for commands outputting client details"""

755

870

all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",

761

876

"Checker", "ExtendedTimeout", "Expires",

762

877

"LastCheckerStatus")

763

878

764

765

879

class DumpJSON(Output):

766

880

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

767

881

data = {properties["Name"]:

768

882

{key: properties[key]

769

883

for key in self.all_keywords}

770

884

for properties in clients.values()}

771

print(json.dumps(data, indent=4, separators=(',', ': ')))

772

885

print(json.dumps(data, indent=4, separators=(",", ": ")))

773

886

774

887

class PrintTable(Output):

775

888

def __init__(self, verbose=False):

783

896

keywords = self.all_keywords

784

897

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

785

898

786

class TableOfClients(object):

899

class TableOfClients:

787

900

tableheaders = {

788

901

"Name": "Name",

789

902

"Enabled": "Enabled",

816

929

817

930

if sys.version_info.major == 2:

818

931

__unicode__ = __str__

932

819

933

def __str__(self):

820

934

return str(self).encode(

821

935

locale.getpreferredencoding())

867

981

minutes=(td.seconds % 3600) // 60,

868

982

seconds=td.seconds % 60))

869

983

870

871

984

class PropertySetter(Base):

872

985

"Abstract class for Actions for setting one client property"

873

986

880

993

def propname(self):

881

994

raise NotImplementedError()

882

995

883

884

996

class Enable(PropertySetter):

885

997

propname = "Enabled"

886

998

value_to_set = True

887

999

888

889

1000

class Disable(PropertySetter):

890

1001

propname = "Enabled"

891

1002

value_to_set = False

892

1003

893

894

1004

class BumpTimeout(PropertySetter):

895

1005

propname = "LastCheckedOK"

896

1006

value_to_set = ""

897

1007

898

899

1008

class StartChecker(PropertySetter):

900

1009

propname = "CheckerRunning"

901

1010

value_to_set = True

902

1011

903

904

1012

class StopChecker(PropertySetter):

905

1013

propname = "CheckerRunning"

906

1014

value_to_set = False

907

1015

908

909

1016

class ApproveByDefault(PropertySetter):

910

1017

propname = "ApprovedByDefault"

911

1018

value_to_set = True

912

1019

913

914

1020

class DenyByDefault(PropertySetter):

915

1021

propname = "ApprovedByDefault"

916

1022

value_to_set = False

917

1023

918

919

1024

class PropertySetterValue(PropertySetter):

920

1025

"""Abstract class for PropertySetter recieving a value as

921

1026

constructor argument instead of a class attribute."""

1027

922

1028

def __init__(self, value):

923

1029

self.value_to_set = value

924

1030

931

1037

class SetChecker(PropertySetterValue):

932

1038

propname = "Checker"

933

1039

934

935

1040

class SetHost(PropertySetterValue):

936

1041

propname = "Host"

937

1042

938

939

1043

class SetSecret(PropertySetterValue):

940

1044

propname = "Secret"

941

1045

949

1053

self._vts = value.read()

950

1054

value.close()

951

1055

952

953

1056

class PropertySetterValueMilliseconds(PropertySetterValue):

954

1057

"""Abstract class for PropertySetterValue taking a value

955

1058

argument as a datetime.timedelta() but should store it as

964

1067

"When setting, convert value from a datetime.timedelta"

965

1068

self._vts = int(round(value.total_seconds() * 1000))

966

1069

967

968

1070

class SetTimeout(PropertySetterValueMilliseconds):

969

1071

propname = "Timeout"

970

1072

971

972

1073

class SetExtendedTimeout(PropertySetterValueMilliseconds):

973

1074

propname = "ExtendedTimeout"

974

1075

975

976

1076

class SetInterval(PropertySetterValueMilliseconds):

977

1077

propname = "Interval"

978

1078

979

980

1079

class SetApprovalDelay(PropertySetterValueMilliseconds):

981

1080

propname = "ApprovalDelay"

982

1081

983

984

1082

class SetApprovalDuration(PropertySetterValueMilliseconds):

985

1083

propname = "ApprovalDuration"

986

1084

1020

1118

"output"))

1021

1119

1022

1120

1023

class Unique(object):

1121

class Unique:

1024

1122

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

1025

1123

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

1026

1124

1310

1408

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1311

1409

1312

1410

def MockDBusPython_func(self, func):

1313

class mock_dbus_python(object):

1411

class mock_dbus_python:

1314

1412

"""mock dbus-python module"""

1315

class exceptions(object):

1413

class exceptions:

1316

1414

"""Pseudo-namespace"""

1317

1415

class DBusException(Exception):

1318

1416

pass

1319

class SystemBus(object):

1417

class SystemBus:

1320

1418

@staticmethod

1321

1419

def get_object(busname, objectpath):

1322

1420

DBusObject = collections.namedtuple(

1336

1434

dbus_interface=dbus_interface)

1337

1435

return DBusObject(methodname=method,

1338

1436

Set=set_property)

1339

class Boolean(object):

1437

class Boolean:

1340

1438

def __init__(self, value):

1341

1439

self.value = bool(value)

1342

1440

def __bool__(self):

1526

1624

finally:

1527

1625

dbus_logger.removeFilter(counting_handler)

1528

1626

1529

self.assertNotIsInstance(e, dbus.ConnectFailed)

1627

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

1530

1628

1531

1629

# Make sure the dbus logger was suppressed

1532

1630

self.assertEqual(0, counting_handler.count)

1554

1652

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

1555

1653

"objectpath", "interface")

1556

1654

1557

class fake_dbus_python_raises_exception_on_connect(object):

1655

class fake_dbus_python_raises_exception_on_connect:

1558

1656

"""fake dbus-python module"""

1559

class exceptions(object):

1657

class exceptions:

1560

1658

"""Pseudo-namespace"""

1561

1659

class DBusException(Exception):

1562

1660

pass

1570

1668

1571

1669

1572

1670

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1573

class mock_dbus_python(object):

1671

class mock_dbus_python:

1574

1672

"""mock dbus-python modules"""

1575

class SystemBus(object):

1673

class SystemBus:

1576

1674

@staticmethod

1577

1675

def get_object(busname, objectpath):

1578

1676

return Unique()

1624

1722

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1625

1723

1626

1724

def Stub_pydbus_func(self, func):

1627

class stub_pydbus(object):

1725

class stub_pydbus:

1628

1726

"""stub pydbus module"""

1629

class SystemBus(object):

1727

class SystemBus:

1630

1728

@staticmethod

1631

1729

def get(busname, objectpath):

1632

1730

DBusObject = collections.namedtuple(

1657

1755

self.assertIs(ret, expected_method_return)

1658

1756

1659

1757

def test_call_method_handles_exception(self):

1660

dbus_logger = logging.getLogger("dbus.proxies")

1661

1662

1758

def func():

1663

1759

raise gi.repository.GLib.Error()

1664

1760

1669

1765

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

1670

1766

"objectpath", "interface")

1671

1767

1672

self.assertNotIsInstance(e, dbus.ConnectFailed)

1768

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

1673

1769

1674

1770

def test_get_converts_to_correct_exception(self):

1675

1771

bus = pydbus_adapter.SystemBus(

1678

1774

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

1679

1775

"objectpath", "interface")

1680

1776

1681

class fake_pydbus_raises_exception_on_connect(object):

1777

class fake_pydbus_raises_exception_on_connect:

1682

1778

"""fake dbus-python module"""

1683

1779

@classmethod

1684

1780

def SystemBus(cls):

1688

1784

return Bus(get=get)

1689

1785

1690

1786

def test_set_property_uses_setattr(self):

1691

class Object(object):

1787

class Object:

1692

1788

pass

1693

1789

obj = Object()

1694

class pydbus_spy(object):

1695

class SystemBus(object):

1790

class pydbus_spy:

1791

class SystemBus:

1696

1792

@staticmethod

1697

1793

def get(busname, objectpath):

1698

1794

return {"interface": obj}

1705

1801

def test_get_suppresses_xml_deprecation_warning(self):

1706

1802

if sys.version_info.major >= 3:

1707

1803

return

1708

class stub_pydbus_get(object):

1709

class SystemBus(object):

1804

class stub_pydbus_get:

1805

class SystemBus:

1710

1806

@staticmethod

1711

1807

def get(busname, objectpath):

1712

1808

warnings.warn_explicit(

1720

1816

1721

1817

1722

1818

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1723

class stub_pydbus(object):

1819

class stub_pydbus:

1724

1820

"""stub pydbus module"""

1725

class SystemBus(object):

1821

class SystemBus:

1726

1822

@staticmethod

1727

1823

def get(busname, objectpath):

1728

1824

return Unique()

1770

1866

self.assertIs(obj1, obj1b)

1771

1867

1772

1868

1869

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1870

1871

class dummy_dbussy:

1872

class DBUS:

1873

class ObjectPath(str):

1874

pass

1875

class DBusError(Exception):

1876

pass

1877

1878

def fake_ravel_func(self, func):

1879

class fake_ravel:

1880

@staticmethod

1881

def system_bus():

1882

class DBusInterfaceProxy:

1883

@staticmethod

1884

def methodname(*args):

1885

return [func(*args)]

1886

class DBusObject:

1887

@staticmethod

1888

def get_interface(interface):

1889

if interface == "interface":

1890

return DBusInterfaceProxy()

1891

return {"busname": {"objectpath": DBusObject()}}

1892

return fake_ravel

1893

1894

def call_method(self, bus, methodname, busname, objectpath,

1895

interface, *args):

1896

with self.assertLogs(log, logging.DEBUG):

1897

return bus.call_method(methodname, busname, objectpath,

1898

interface, *args)

1899

1900

def test_call_method_returns(self):

1901

expected_method_return = Unique()

1902

method_args = (Unique(), Unique())

1903

def func(*args):

1904

self.assertEqual(len(method_args), len(args))

1905

for marg, arg in zip(method_args, args):

1906

self.assertIs(marg, arg)

1907

return expected_method_return

1908

fake_ravel = self.fake_ravel_func(func)

1909

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1910

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

1911

"objectpath", "interface",

1912

*method_args)

1913

self.assertIs(ret, expected_method_return)

1914

1915

def test_call_method_filters_objectpath(self):

1916

def func():

1917

return method_return

1918

fake_ravel = self.fake_ravel_func(func)

1919

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1920

method_return = (self.dummy_dbussy.DBUS

1921

.ObjectPath("objectpath"))

1922

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

1923

"objectpath", "interface")

1924

self.assertEqual("objectpath", ret)

1925

self.assertNotIsInstance(ret,

1926

self.dummy_dbussy.DBUS.ObjectPath)

1927

1928

def test_call_method_filters_objectpaths_in_dict(self):

1929

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1930

def func():

1931

return method_return

1932

fake_ravel = self.fake_ravel_func(func)

1933

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1934

method_return = {

1935

ObjectPath("objectpath_key_1"):

1936

ObjectPath("objectpath_value_1"),

1937

ObjectPath("objectpath_key_2"):

1938

ObjectPath("objectpath_value_2"),

1939

}

1940

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

1941

"objectpath", "interface")

1942

expected_method_return = {str(key): str(value)

1943

for key, value in

1944

method_return.items()}

1945

for key, value in ret.items():

1946

self.assertNotIsInstance(key, ObjectPath)

1947

self.assertNotIsInstance(value, ObjectPath)

1948

self.assertEqual(expected_method_return, ret)

1949

self.assertIsInstance(ret, dict)

1950

1951

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1952

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1953

def func():

1954

return method_return

1955

fake_ravel = self.fake_ravel_func(func)

1956

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1957

method_return = {

1958

ObjectPath("key1"): {

1959

ObjectPath("key11"): ObjectPath("value11"),

1960

ObjectPath("key12"): ObjectPath("value12"),

1961

},

1962

ObjectPath("key2"): {

1963

ObjectPath("key21"): ObjectPath("value21"),

1964

ObjectPath("key22"): ObjectPath("value22"),

1965

},

1966

}

1967

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

1968

"objectpath", "interface")

1969

expected_method_return = {

1970

"key1": {"key11": "value11",

1971

"key12": "value12"},

1972

"key2": {"key21": "value21",

1973

"key22": "value22"},

1974

}

1975

self.assertEqual(expected_method_return, ret)

1976

for key, value in ret.items():

1977

self.assertIsInstance(value, dict)

1978

self.assertEqual(expected_method_return[key], value)

1979

self.assertNotIsInstance(key, ObjectPath)

1980

for inner_key, inner_value in value.items():

1981

self.assertIsInstance(value, dict)

1982

self.assertEqual(

1983

expected_method_return[key][inner_key],

1984

inner_value)

1985

self.assertNotIsInstance(key, ObjectPath)

1986

1987

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

1988

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1989

def func():

1990

return method_return

1991

fake_ravel = self.fake_ravel_func(func)

1992

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1993

method_return = {

1994

ObjectPath("key1"): {

1995

ObjectPath("key2"): {

1996

ObjectPath("key3"): ObjectPath("value"),

1997

},

1998

},

1999

}

2000

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

2001

"objectpath", "interface")

2002

expected_method_return = {"key1": {"key2": {"key3": "value"}}}

2003

self.assertEqual(expected_method_return, ret)

2004

self.assertIsInstance(ret, dict)

2005

self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)

2006

self.assertIsInstance(ret["key1"], dict)

2007

self.assertNotIsInstance(next(iter(ret["key1"].keys())),

2008

ObjectPath)

2009

self.assertIsInstance(ret["key1"]["key2"], dict)

2010

self.assertNotIsInstance(

2011

next(iter(ret["key1"]["key2"].keys())),

2012

ObjectPath)

2013

self.assertEqual("value", ret["key1"]["key2"]["key3"])

2014

self.assertNotIsInstance(ret["key1"]["key2"]["key3"],

2015

self.dummy_dbussy.DBUS.ObjectPath)

2016

2017

def test_call_method_handles_exception(self):

2018

def func():

2019

raise self.dummy_dbussy.DBusError()

2020

2021

fake_ravel = self.fake_ravel_func(func)

2022

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

2023

2024

with self.assertRaises(dbus.Error) as e:

2025

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

2026

"objectpath", "interface")

2027

2028

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

2029

2030

def test_get_object_converts_to_correct_exception(self):

2031

class fake_ravel_raises_exception_on_connect:

2032

@staticmethod

2033

def system_bus():

2034

class Bus:

2035

@staticmethod

2036

def __getitem__(key):

2037

if key == "objectpath":

2038

raise self.dummy_dbussy.DBusError()

2039

raise Exception(key)

2040

return {"busname": Bus()}

2041

def func():

2042

raise self.dummy_dbussy.DBusError()

2043

bus = dbussy_adapter.SystemBus(

2044

self.dummy_dbussy,

2045

fake_ravel_raises_exception_on_connect)

2046

with self.assertRaises(dbus.ConnectFailed):

2047

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

2048

"objectpath", "interface")

2049

2050

1773

2051

class Test_commands_from_options(unittest.TestCase):

1774

2052

1775

2053

def setUp(self):

2001

2279

2002

2280

2003

2281

def test_manual_page_example_1(self):

2004

self.assert_command_from_args("--verbose".split(),

2282

self.assert_command_from_args("",

2005

2283

command.PrintTable,

2006

2284

clients=[],

2007

verbose=True)

2285

verbose=False)

2008

2286

2009

2287

def test_manual_page_example_2(self):

2010

2288

self.assert_command_from_args(

2149

2427

busname = "se.recompile.Mandos"

2150

2428

client_interface = "se.recompile.Mandos.Client"

2151

2429

command.Approve().run(self.bus.clients, self.bus)

2430

self.assertTrue(self.bus.clients)

2152

2431

for clientpath in self.bus.clients:

2153

2432

self.assertIn(("Approve", busname, clientpath,

2154

2433

client_interface, (True,)), self.bus.calls)

2157

2436

busname = "se.recompile.Mandos"

2158

2437

client_interface = "se.recompile.Mandos.Client"

2159

2438

command.Deny().run(self.bus.clients, self.bus)

2439

self.assertTrue(self.bus.clients)

2160

2440

for clientpath in self.bus.clients:

2161

2441

self.assertIn(("Approve", busname, clientpath,

2162

2442

client_interface, (False,)),

2163

2443

self.bus.calls)

2164

2444

2165

2445

def test_Remove(self):

2446

busname = "se.recompile.Mandos"

2447

server_path = "/"

2448

server_interface = "se.recompile.Mandos"

2449

orig_clients = self.bus.clients.copy()

2166

2450

command.Remove().run(self.bus.clients, self.bus)

2167

for clientpath in self.bus.clients:

2168

self.assertIn(("RemoveClient", dbus_busname,

2169

dbus_server_path, dbus_server_interface,

2451

self.assertFalse(self.bus.clients)

2452

for clientpath in orig_clients:

2453

self.assertIn(("RemoveClient", busname,

2454

server_path, server_interface,

2170

2455

(clientpath,)), self.bus.calls)

2171

2456

2172

2457

expected_json = {

2374

2659

else:

2375

2660

cmd_args = [() for x in range(len(self.values_to_get))]

2376

2661

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2377

2663

for value_to_get, cmd_arg in zip(values_to_get, cmd_args):

2378

2664

for clientpath in self.bus.clients:

2379

2665

self.bus.clients[clientpath][self.propname] = (

2380

2666

Unique())

2381

2667

self.command(*cmd_arg).run(self.bus.clients, self.bus)

2668

self.assertTrue(self.bus.clients)

2382

2669

for clientpath in self.bus.clients:

2383

2670

value = (self.bus.clients[clientpath]

2384

2671

[self.propname])

2443

2730

class TestSetSecretCmd(TestPropertySetterCmd):

2444

2731

command = command.SetSecret

2445

2732

propname = "Secret"

2446

values_to_set = [io.BytesIO(b""),

2447

io.BytesIO(b"secret\0xyzzy\nbar")]

2448

values_to_get = [f.getvalue() for f in values_to_set]

2733

def __init__(self, *args, **kwargs):

2734

self.values_to_set = [io.BytesIO(b""),

2735

io.BytesIO(b"secret\0xyzzy\nbar")]

2736

self.values_to_get = [f.getvalue() for f in

2737

self.values_to_set]

2738

super(TestSetSecretCmd, self).__init__(*args, **kwargs)

2449

2739

2450

2740

2451

2741

class TestSetTimeoutCmd(TestPropertySetterCmd):

2504

2794

2505

2795

2506

2796

2507

def should_only_run_tests():

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2508

2799

parser = argparse.ArgumentParser(add_help=False)

2509

parser.add_argument("--check", action='store_true')

2800

parser.add_argument("--check", action="store_true")

2801

parser.add_argument("--prefix", )

2510

2802

args, unknown_args = parser.parse_known_args()

2511

run_tests = args.check

2512

if run_tests:

2513

# Remove --check argument from sys.argv

2803

if args.check:

2804

# Remove test options from sys.argv

2514

2805

sys.argv[1:] = unknown_args

2515

return run_tests

2806

return args

2516

2807

2517

2808

# Add all tests from doctest strings

2518

2809

def load_tests(loader, tests, none):

2521

2812

return tests

2522

2813

2523

2814

if __name__ == "__main__":

2815

options = parse_test_args()

2524

2816

try:

2525

if should_only_run_tests():

2526

# Call using ./tdd-python-script --check [--verbose]

2527

unittest.main()

2817

if options.check:

2818

extra_test_prefix = options.prefix

2819

if extra_test_prefix is not None:

2820

if not (unittest.main(argv=[""], exit=False)

2821

.result.wasSuccessful()):

2822

sys.exit(1)

2823

class ExtraTestLoader(unittest.TestLoader):

2824

testMethodPrefix = extra_test_prefix

2825

# Call using ./scriptname --check [--verbose]

2826

unittest.main(argv=[""], testLoader=ExtraTestLoader())

2827

else:

2828

unittest.main(argv=[""])

2528

2829

else:

2529

2830

main()

2530

2831

finally:

2531

2832

logging.shutdown()

2833

2834

# Local Variables:

2835

# run-tests:

2836

# (lambda (&optional extra)

2837

# (if (not (funcall run-tests-in-test-buffer default-directory

2838

# extra))

2839

# (funcall show-test-buffer-in-test-window)

2840

# (funcall remove-test-window)

2841

# (if extra (message "Extra tests run successfully!"))))

2842

# run-tests-in-test-buffer:

2843

# (lambda (dir &optional extra)

2844

# (with-current-buffer (get-buffer-create "*Test*")

2845

# (setq buffer-read-only nil

2846

# default-directory dir)

2847

# (erase-buffer)

2848

# (compilation-mode))

2849

# (let ((process-result

2850

# (let ((inhibit-read-only t))

2851

# (process-file-shell-command

2852

# (funcall get-command-line extra) nil "*Test*"))))

2853

# (and (numberp process-result)

2854

# (= process-result 0))))

2855

# get-command-line:

2856

# (lambda (&optional extra)

2857

# (let ((quoted-script

2858

# (shell-quote-argument (funcall get-script-name))))

2859

# (format

2860

# (concat "%s --check" (if extra " --prefix=atest" ""))

2861

# quoted-script)))

2862

# get-script-name:

2863

# (lambda ()

2864

# (if (fboundp 'file-local-name)

2865

# (file-local-name (buffer-file-name))

2866

# (or (file-remote-p (buffer-file-name) 'localname)

2867

# (buffer-file-name))))

2868

# remove-test-window:

2869

# (lambda ()

2870

# (let ((test-window (get-buffer-window "*Test*")))

2871

# (if test-window (delete-window test-window))))

2872

# show-test-buffer-in-test-window:

2873

# (lambda ()

2874

# (when (not (get-buffer-window-list "*Test*"))

2875

# (setq next-error-last-buffer (get-buffer "*Test*"))

2876

# (let* ((side (if (>= (window-width) 146) 'right 'bottom))

2877

# (display-buffer-overriding-action

2878

# `((display-buffer-in-side-window) (side . ,side)

2879

# (window-height . fit-window-to-buffer)

2880

# (window-width . fit-window-to-buffer))))

2881

# (display-buffer "*Test*"))))

2882

# eval:

2883

# (progn

2884

# (let* ((run-extra-tests (lambda () (interactive)

2885

# (funcall run-tests t)))

2886

# (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t

2887

# (outer-keymap `(keymap (3 . ,inner-keymap)))) ; C-c

2888

# (setq minor-mode-overriding-map-alist

2889

# (cons `(run-tests . ,outer-keymap)

2890

# minor-mode-overriding-map-alist)))

2891

# (add-hook 'after-save-hook run-tests 90 t))

2892

# End: