/mandos/release : revision 237.7.634

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

Committer: Teddy Hogeborn
Date: 2019-03-18 22:29:25 UTC
mto: This revision was merged to the branch mainline in revision 382.
Revision ID: teddy@recompile.se-20190318222925-jvhek84dgcfgj6g3

mandos-ctl: Refactor tests

* mandos-ctl: Where the clients names "foo" and "barbar" do not refer
              to the actual mock clients in the TestCommand class,
              change all occurrences of these names to "client1" and
              "client2" (or just "client" when only one is used) .
              Also change all test doubles to use correct terminology;
              some things called mocks are actually stubs or spies,
              and rename all true mocks to have "mock" in their names.
              Also eliminate duplicate values in tests; derive values
              from previously defined values whenever possible.

files modified:
intro.xml

mandos

mandos-ctl

plugin-runner.c

Show diffs side-by-side

added added

removed removed

mandos-ctl

import tempfile

import contextlib

try:

import pydbus

import gi

dbus_python = None

except ImportError:

import dbus as dbus_python

pydbus = None

class gi(object):

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

class repository(object):

class GLib(object):

class Error(Exception):

pass

import dbus

# Show warnings by default

if not sys.warnoptions:

locale.setlocale(locale.LC_ALL, "")

dbus_busname_domain = "se.recompile"

dbus_busname = dbus_busname_domain + ".Mandos"

server_dbus_path = "/"

server_dbus_interface = dbus_busname_domain + ".Mandos"

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

del dbus_busname_domain

version = "1.8.3"

try:

dbus.OBJECT_MANAGER_IFACE

except AttributeError:

dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"

def main():

parser = argparse.ArgumentParser()

add_command_line_options(parser)

options = parser.parse_args()

check_option_syntax(parser, options)

clientnames = options.client

if options.debug:

log.setLevel(logging.DEBUG)

if pydbus is not None:

bus = pydbus_adapter.CachingBus(pydbus)

else:

bus = dbus_python_adapter.CachingBus(dbus_python)

100

101

try:

102

all_clients = bus.get_clients_and_properties()

103

except dbus.ConnectFailed as e:

104

log.critical("Could not connect to Mandos server: %s", e)

105

sys.exit(1)

106

except dbus.Error as e:

107

log.critical(

108

"Failed to access Mandos server through D-Bus:\n%s", e)

109

sys.exit(1)

bus = dbus.SystemBus()

100

mandos_dbus_object = get_mandos_dbus_object(bus)

101

102

mandos_serv = dbus.Interface(

103

mandos_dbus_object, dbus_interface=server_dbus_interface)

104

mandos_serv_object_manager = dbus.Interface(

105

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

106

107

managed_objects = get_managed_objects(mandos_serv_object_manager)

108

109

all_clients = {}

110

for path, ifs_and_props in managed_objects.items():

111

try:

112

all_clients[path] = ifs_and_props[client_dbus_interface]

113

except KeyError:

114

pass

110

115

111

116

# Compile dict of (clientpath: properties) to process

112

117

if not clientnames:

125

130

commands = commands_from_options(options)

126

131

127

132

for command in commands:

128

command.run(clients, bus)

133

command.run(clients, bus, mandos_serv)

129

134

130

135

131

136

def add_command_line_options(parser):

136

141

help="Select all clients")

137

142

parser.add_argument("-v", "--verbose", action="store_true",

138

143

help="Print all fields")

139

parser.add_argument("-j", "--dump-json", dest="commands",

140

action="append_const", default=[],

141

const=command.DumpJSON(),

144

parser.add_argument("-j", "--dump-json", action="store_true",

142

145

help="Dump client data in JSON format")

143

146

enable_disable = parser.add_mutually_exclusive_group()

144

enable_disable.add_argument("-e", "--enable", dest="commands",

145

action="append_const", default=[],

146

const=command.Enable(),

147

enable_disable.add_argument("-e", "--enable", action="store_true",

147

148

help="Enable client")

148

enable_disable.add_argument("-d", "--disable", dest="commands",

149

action="append_const", default=[],

150

const=command.Disable(),

149

enable_disable.add_argument("-d", "--disable",

150

action="store_true",

151

help="disable client")

152

parser.add_argument("-b", "--bump-timeout", dest="commands",

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

152

parser.add_argument("-b", "--bump-timeout", action="store_true",

155

153

help="Bump timeout for client")

156

154

start_stop_checker = parser.add_mutually_exclusive_group()

157

155

start_stop_checker.add_argument("--start-checker",

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

156

action="store_true",

161

157

help="Start checker for client")

162

start_stop_checker.add_argument("--stop-checker", dest="commands",

163

action="append_const", default=[],

164

const=command.StopChecker(),

158

start_stop_checker.add_argument("--stop-checker",

159

action="store_true",

165

160

help="Stop checker for client")

166

parser.add_argument("-V", "--is-enabled", dest="commands",

167

action="append_const", default=[],

168

const=command.IsEnabled(),

161

parser.add_argument("-V", "--is-enabled", action="store_true",

169

162

help="Check if client is enabled")

170

parser.add_argument("-r", "--remove", dest="commands",

171

action="append_const", default=[],

172

const=command.Remove(),

163

parser.add_argument("-r", "--remove", action="store_true",

173

164

help="Remove client")

174

parser.add_argument("-c", "--checker", dest="commands",

175

action="append", default=[],

176

metavar="COMMAND", type=command.SetChecker,

165

parser.add_argument("-c", "--checker",

177

166

help="Set checker command for client")

178

parser.add_argument(

179

"-t", "--timeout", dest="commands", action="append",

180

default=[], metavar="TIME",

181

type=command.SetTimeout.argparse(string_to_delta),

182

help="Set timeout for client")

183

parser.add_argument(

184

"--extended-timeout", dest="commands", action="append",

185

default=[], metavar="TIME",

186

type=command.SetExtendedTimeout.argparse(string_to_delta),

187

help="Set extended timeout for client")

188

parser.add_argument(

189

"-i", "--interval", dest="commands", action="append",

190

default=[], metavar="TIME",

191

type=command.SetInterval.argparse(string_to_delta),

192

help="Set checker interval for client")

167

parser.add_argument("-t", "--timeout", type=string_to_delta,

168

help="Set timeout for client")

169

parser.add_argument("--extended-timeout", type=string_to_delta,

170

help="Set extended timeout for client")

171

parser.add_argument("-i", "--interval", type=string_to_delta,

172

help="Set checker interval for client")

193

173

approve_deny_default = parser.add_mutually_exclusive_group()

194

174

approve_deny_default.add_argument(

195

"--approve-by-default", dest="commands",

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

175

"--approve-by-default", action="store_true",

176

default=None, dest="approved_by_default",

198

177

help="Set client to be approved by default")

199

178

approve_deny_default.add_argument(

200

"--deny-by-default", dest="commands",

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

179

"--deny-by-default", action="store_false",

180

dest="approved_by_default",

203

181

help="Set client to be denied by default")

204

parser.add_argument(

205

"--approval-delay", dest="commands", action="append",

206

default=[], metavar="TIME",

207

type=command.SetApprovalDelay.argparse(string_to_delta),

208

help="Set delay before client approve/deny")

209

parser.add_argument(

210

"--approval-duration", dest="commands", action="append",

211

default=[], metavar="TIME",

212

type=command.SetApprovalDuration.argparse(string_to_delta),

213

help="Set duration of one client approval")

214

parser.add_argument("-H", "--host", dest="commands",

215

action="append", default=[], metavar="STRING",

216

type=command.SetHost,

217

help="Set host for client")

218

parser.add_argument(

219

"-s", "--secret", dest="commands", action="append",

220

default=[], metavar="FILENAME",

221

type=command.SetSecret.argparse(argparse.FileType(mode="rb")),

222

help="Set password blob (file) for client")

182

parser.add_argument("--approval-delay", type=string_to_delta,

183

help="Set delay before client approve/deny")

184

parser.add_argument("--approval-duration", type=string_to_delta,

185

help="Set duration of one client approval")

186

parser.add_argument("-H", "--host", help="Set host for client")

187

parser.add_argument("-s", "--secret",

188

type=argparse.FileType(mode="rb"),

189

help="Set password blob (file) for client")

223

190

approve_deny = parser.add_mutually_exclusive_group()

224

191

approve_deny.add_argument(

225

"-A", "--approve", dest="commands", action="append_const",

226

default=[], const=command.Approve(),

192

"-A", "--approve", action="store_true",

227

193

help="Approve any current client request")

228

approve_deny.add_argument("-D", "--deny", dest="commands",

229

action="append_const", default=[],

230

const=command.Deny(),

194

approve_deny.add_argument("-D", "--deny", action="store_true",

231

195

help="Deny any current client request")

232

196

parser.add_argument("--debug", action="store_true",

233

197

help="Debug mode (show D-Bus commands)")

424

388

"""Apply additional restrictions on options, not expressible in

425

389

argparse"""

426

390

427

def has_commands(options, commands=None):

428

if commands is None:

429

commands = (command.Enable,

430

command.Disable,

431

command.BumpTimeout,

432

command.StartChecker,

433

command.StopChecker,

434

command.IsEnabled,

435

command.Remove,

436

command.SetChecker,

437

command.SetTimeout,

438

command.SetExtendedTimeout,

439

command.SetInterval,

440

command.ApproveByDefault,

441

command.DenyByDefault,

442

command.SetApprovalDelay,

443

command.SetApprovalDuration,

444

command.SetHost,

445

command.SetSecret,

446

command.Approve,

447

command.Deny)

448

return any(isinstance(cmd, commands)

449

for cmd in options.commands)

391

def has_actions(options):

392

return any((options.enable,

393

options.disable,

394

options.bump_timeout,

395

options.start_checker,

396

options.stop_checker,

397

options.is_enabled,

398

options.remove,

399

options.checker is not None,

400

options.timeout is not None,

401

options.extended_timeout is not None,

402

options.interval is not None,

403

options.approved_by_default is not None,

404

options.approval_delay is not None,

405

options.approval_duration is not None,

406

options.host is not None,

407

options.secret is not None,

408

options.approve,

409

options.deny))

450

410

451

if has_commands(options) and not (options.client or options.all):

411

if has_actions(options) and not (options.client or options.all):

452

412

parser.error("Options require clients names or --all.")

453

if options.verbose and has_commands(options):

413

if options.verbose and has_actions(options):

454

414

parser.error("--verbose can only be used alone.")

455

if (has_commands(options, (command.DumpJSON,))

456

and (options.verbose or len(options.commands) > 1)):

415

if options.dump_json and (options.verbose

416

or has_actions(options)):

457

417

parser.error("--dump-json can only be used alone.")

458

if options.all and not has_commands(options):

418

if options.all and not has_actions(options):

459

419

parser.error("--all requires an action.")

460

if (has_commands(options, (command.IsEnabled,))

461

and len(options.client) > 1):

420

if options.is_enabled and len(options.client) > 1:

462

421

parser.error("--is-enabled requires exactly one client")

463

if (len(options.commands) > 1

464

and has_commands(options, (command.Remove,))

465

and not has_commands(options, (command.Deny,))):

466

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

467

468

469

class dbus(object):

470

471

class SystemBus(object):

472

473

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

474

def get_managed_objects(self, busname, objectpath):

475

return self.call_method("GetManagedObjects", busname,

476

objectpath,

477

self.object_manager_iface)

478

479

properties_iface = "org.freedesktop.DBus.Properties"

480

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

481

value):

482

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

483

self.properties_iface, interface, key,

484

value)

485

486

487

class MandosBus(SystemBus):

488

busname_domain = "se.recompile"

489

busname = busname_domain + ".Mandos"

490

server_path = "/"

491

server_interface = busname_domain + ".Mandos"

492

client_interface = busname_domain + ".Mandos.Client"

493

del busname_domain

494

495

def get_clients_and_properties(self):

496

managed_objects = self.get_managed_objects(

497

self.busname, self.server_path)

498

return {objpath: properties[self.client_interface]

499

for objpath, properties in managed_objects.items()

500

if self.client_interface in properties}

501

502

def set_client_property(self, objectpath, key, value):

503

return self.set_property(self.busname, objectpath,

504

self.client_interface, key,

505

value)

506

507

def call_client_method(self, objectpath, method, *args):

508

return self.call_method(method, self.busname, objectpath,

509

self.client_interface, *args)

510

511

def call_server_method(self, method, *args):

512

return self.call_method(method, self.busname,

513

self.server_path,

514

self.server_interface, *args)

515

516

class Error(Exception):

517

pass

518

519

class ConnectFailed(Error):

520

pass

521

522

523

class dbus_python_adapter(object):

524

525

class SystemBus(dbus.MandosBus):

526

"""Use dbus-python"""

527

528

def __init__(self, module=dbus_python):

529

self.dbus_python = module

530

self.bus = self.dbus_python.SystemBus()

531

532

@contextlib.contextmanager

533

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

534

try:

535

yield

536

except self.dbus_python.exceptions.DBusException as e:

537

# This does what "raise from" would do

538

exc = exception_class(*e.args)

539

exc.__cause__ = e

540

raise exc

541

542

def call_method(self, methodname, busname, objectpath,

543

interface, *args):

544

proxy_object = self.get_object(busname, objectpath)

545

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

546

interface, methodname,

547

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

548

method = getattr(proxy_object, methodname)

549

with self.convert_exception():

550

with dbus_python_adapter.SilenceLogger(

551

"dbus.proxies"):

552

value = method(*args, dbus_interface=interface)

553

return self.type_filter(value)

554

555

def get_object(self, busname, objectpath):

556

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

557

busname, objectpath)

558

with self.convert_exception(dbus.ConnectFailed):

559

return self.bus.get_object(busname, objectpath)

560

561

def type_filter(self, value):

562

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

563

if isinstance(value, self.dbus_python.Boolean):

564

return bool(value)

565

if isinstance(value, self.dbus_python.ObjectPath):

566

return str(value)

567

# Also recurse into dictionaries

568

if isinstance(value, self.dbus_python.Dictionary):

569

return {self.type_filter(key):

570

self.type_filter(subval)

571

for key, subval in value.items()}

572

return value

573

574

575

class SilenceLogger(object):

576

"Simple context manager to silence a particular logger"

577

def __init__(self, loggername):

578

self.logger = logging.getLogger(loggername)

579

580

def __enter__(self):

581

self.logger.addFilter(self.nullfilter)

582

583

class NullFilter(logging.Filter):

584

def filter(self, record):

585

return False

586

587

nullfilter = NullFilter()

588

589

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

590

self.logger.removeFilter(self.nullfilter)

591

592

593

class CachingBus(SystemBus):

594

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

595

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

596

self.object_cache = {}

597

super(dbus_python_adapter.CachingBus,

598

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

599

def get_object(self, busname, objectpath):

600

try:

601

return self.object_cache[(busname, objectpath)]

602

except KeyError:

603

new_object = super(

604

dbus_python_adapter.CachingBus,

605

self).get_object(busname, objectpath)

606

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

607

return new_object

608

609

610

class pydbus_adapter(object):

611

class SystemBus(dbus.MandosBus):

612

def __init__(self, module=pydbus):

613

self.pydbus = module

614

self.bus = self.pydbus.SystemBus()

615

616

@contextlib.contextmanager

617

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

618

try:

619

yield

620

except gi.repository.GLib.Error as e:

621

# This does what "raise from" would do

622

exc = exception_class(*e.args)

623

exc.__cause__ = e

624

raise exc

625

626

def call_method(self, methodname, busname, objectpath,

627

interface, *args):

628

proxy_object = self.get(busname, objectpath)

629

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

630

interface, methodname,

631

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

632

method = getattr(proxy_object[interface], methodname)

633

with self.convert_exception():

634

return method(*args)

635

636

def get(self, busname, objectpath):

637

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

638

busname, objectpath)

639

with self.convert_exception(dbus.ConnectFailed):

640

if sys.version_info.major <= 2:

641

with warnings.catch_warnings():

642

warnings.filterwarnings(

643

"ignore", "", DeprecationWarning,

644

r"^xml\.etree\.ElementTree$")

645

return self.bus.get(busname, objectpath)

646

else:

647

return self.bus.get(busname, objectpath)

648

649

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

650

value):

651

proxy_object = self.get(busname, objectpath)

652

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

653

objectpath, self.properties_iface, interface,

654

key, value)

655

setattr(proxy_object[interface], key, value)

656

657

class CachingBus(SystemBus):

658

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

659

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

660

self.object_cache = {}

661

super(pydbus_adapter.CachingBus,

662

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

663

def get(self, busname, objectpath):

664

try:

665

return self.object_cache[(busname, objectpath)]

666

except KeyError:

667

new_object = (super(pydbus_adapter.CachingBus, self)

668

.get(busname, objectpath))

669

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

670

return new_object

422

if options.remove:

423

options.remove = False

424

if has_actions(options) and not options.deny:

425

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

426

options.remove = True

427

428

429

def get_mandos_dbus_object(bus):

430

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

431

dbus_busname, server_dbus_path)

432

with if_dbus_exception_log_with_exception_and_exit(

433

"Could not connect to Mandos server: %s"):

434

mandos_dbus_object = bus.get_object(dbus_busname,

435

server_dbus_path)

436

return mandos_dbus_object

437

438

439

@contextlib.contextmanager

440

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

441

try:

442

yield

443

except dbus.exceptions.DBusException as e:

444

log.critical(*(args + (e,)), **kwargs)

445

sys.exit(1)

446

447

448

def get_managed_objects(object_manager):

449

log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,

450

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

451

with if_dbus_exception_log_with_exception_and_exit(

452

"Failed to access Mandos server through D-Bus:\n%s"):

453

with SilenceLogger("dbus.proxies"):

454

managed_objects = object_manager.GetManagedObjects()

455

return managed_objects

456

457

458

class SilenceLogger(object):

459

"Simple context manager to silence a particular logger"

460

def __init__(self, loggername):

461

self.logger = logging.getLogger(loggername)

462

463

def __enter__(self):

464

self.logger.addFilter(self.nullfilter)

465

return self

466

467

class NullFilter(logging.Filter):

468

def filter(self, record):

469

return False

470

471

nullfilter = NullFilter()

472

473

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

474

self.logger.removeFilter(self.nullfilter)

671

475

672

476

673

477

def commands_from_options(options):

674

478

675

commands = list(options.commands)

676

677

def find_cmd(cmd, commands):

678

i = 0

679

for i, c in enumerate(commands):

680

if isinstance(c, cmd):

681

return i

682

return i+1

683

684

# If command.Remove is present, move any instances of command.Deny

685

# to occur ahead of command.Remove.

686

index_of_remove = find_cmd(command.Remove, commands)

687

before_remove = commands[:index_of_remove]

688

after_remove = commands[index_of_remove:]

689

cleaned_after = []

690

for cmd in after_remove:

691

if isinstance(cmd, command.Deny):

692

before_remove.append(cmd)

479

commands = []

480

481

if options.is_enabled:

482

commands.append(command.IsEnabled())

483

484

if options.approve:

485

commands.append(command.Approve())

486

487

if options.deny:

488

commands.append(command.Deny())

489

490

if options.remove:

491

commands.append(command.Remove())

492

493

if options.dump_json:

494

commands.append(command.DumpJSON())

495

496

if options.enable:

497

commands.append(command.Enable())

498

499

if options.disable:

500

commands.append(command.Disable())

501

502

if options.bump_timeout:

503

commands.append(command.BumpTimeout())

504

505

if options.start_checker:

506

commands.append(command.StartChecker())

507

508

if options.stop_checker:

509

commands.append(command.StopChecker())

510

511

if options.approved_by_default is not None:

512

if options.approved_by_default:

513

commands.append(command.ApproveByDefault())

693

514

else:

694

cleaned_after.append(cmd)

695

if cleaned_after != after_remove:

696

commands = before_remove + cleaned_after

515

commands.append(command.DenyByDefault())

516

517

if options.checker is not None:

518

commands.append(command.SetChecker(options.checker))

519

520

if options.host is not None:

521

commands.append(command.SetHost(options.host))

522

523

if options.secret is not None:

524

commands.append(command.SetSecret(options.secret))

525

526

if options.timeout is not None:

527

commands.append(command.SetTimeout(options.timeout))

528

529

if options.extended_timeout:

530

commands.append(

531

command.SetExtendedTimeout(options.extended_timeout))

532

533

if options.interval is not None:

534

commands.append(command.SetInterval(options.interval))

535

536

if options.approval_delay is not None:

537

commands.append(

538

command.SetApprovalDelay(options.approval_delay))

539

540

if options.approval_duration is not None:

541

commands.append(

542

command.SetApprovalDuration(options.approval_duration))

697

543

698

544

# If no command option has been given, show table of clients,

699

545

# optionally verbosely

708

554

709

555

class Base(object):

710

556

"""Abstract base class for commands"""

711

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

557

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

712

558

"""Normal commands should implement run_on_one_client(),

713

559

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

714

560

override this run() method instead.

715

561

"""

716

self.bus = bus

717

for client, properties in clients.items():

562

self.mandos = mandos

563

for clientpath, properties in clients.items():

564

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

565

dbus_busname, str(clientpath))

566

client = bus.get_object(dbus_busname, clientpath)

718

567

self.run_on_one_client(client, properties)

719

568

720

569

721

570

class IsEnabled(Base):

722

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

723

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

724

if properties["Enabled"]:

571

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

572

client, properties = next(iter(clients.items()))

573

if self.is_enabled(client, properties):

725

574

sys.exit(0)

726

575

sys.exit(1)

576

def is_enabled(self, client, properties):

577

return properties["Enabled"]

727

578

728

579

729

580

class Approve(Base):

730

581

def run_on_one_client(self, client, properties):

731

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

582

log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,

583

client.__dbus_object_path__,

584

client_dbus_interface)

585

client.Approve(dbus.Boolean(True),

586

dbus_interface=client_dbus_interface)

732

587

733

588

734

589

class Deny(Base):

735

590

def run_on_one_client(self, client, properties):

736

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

591

log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,

592

client.__dbus_object_path__,

593

client_dbus_interface)

594

client.Approve(dbus.Boolean(False),

595

dbus_interface=client_dbus_interface)

737

596

738

597

739

598

class Remove(Base):

740

def run(self, clients, bus):

741

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

742

bus.call_server_method("RemoveClient", clientpath)

599

def run_on_one_client(self, client, properties):

600

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

601

dbus_busname, server_dbus_path,

602

server_dbus_interface,

603

str(client.__dbus_object_path__))

604

self.mandos.RemoveClient(client.__dbus_object_path__)

743

605

744

606

745

607

class Output(Base):

755

617

756

618

757

619

class DumpJSON(Output):

758

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

759

data = {properties["Name"]:

760

{key: properties[key]

620

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

621

data = {client["Name"]:

622

{key: self.dbus_boolean_to_bool(client[key])

761

623

for key in self.all_keywords}

762

for properties in clients.values()}

624

for client in clients.values()}

763

625

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

764

626

627

@staticmethod

628

def dbus_boolean_to_bool(value):

629

if isinstance(value, dbus.Boolean):

630

value = bool(value)

631

return value

632

765

633

766

634

class PrintTable(Output):

767

635

def __init__(self, verbose=False):

768

636

self.verbose = verbose

769

637

770

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

638

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

771

639

default_keywords = ("Name", "Enabled", "Timeout",

772

640

"LastCheckedOK")

773

641

keywords = default_keywords

834

702

835

703

@classmethod

836

704

def valuetostring(cls, value, keyword):

837

if isinstance(value, bool):

705

if isinstance(value, dbus.Boolean):

838

706

return "Yes" if value else "No"

839

707

if keyword in ("Timeout", "Interval", "ApprovalDelay",

840

708

"ApprovalDuration", "ExtendedTimeout"):

860

728

seconds=td.seconds % 60))

861

729

862

730

863

class PropertySetter(Base):

731

class Property(Base):

864

732

"Abstract class for Actions for setting one client property"

865

733

866

def run_on_one_client(self, client, properties=None):

734

def run_on_one_client(self, client, properties):

867

735

"""Set the Client's D-Bus property"""

868

self.bus.set_client_property(client, self.propname,

869

self.value_to_set)

736

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

737

client.__dbus_object_path__,

738

dbus.PROPERTIES_IFACE, client_dbus_interface,

739

self.propname, self.value_to_set

740

if not isinstance(self.value_to_set,

741

dbus.Boolean)

742

else bool(self.value_to_set))

743

client.Set(client_dbus_interface, self.propname,

744

self.value_to_set,

745

dbus_interface=dbus.PROPERTIES_IFACE)

870

746

871

747

@property

872

748

def propname(self):

873

749

raise NotImplementedError()

874

750

875

751

876

class Enable(PropertySetter):

877

propname = "Enabled"

878

value_to_set = True

879

880

881

class Disable(PropertySetter):

882

propname = "Enabled"

883

value_to_set = False

884

885

886

class BumpTimeout(PropertySetter):

752

class Enable(Property):

753

propname = "Enabled"

754

value_to_set = dbus.Boolean(True)

755

756

757

class Disable(Property):

758

propname = "Enabled"

759

value_to_set = dbus.Boolean(False)

760

761

762

class BumpTimeout(Property):

887

763

propname = "LastCheckedOK"

888

764

value_to_set = ""

889

765

890

766

891

class StartChecker(PropertySetter):

892

propname = "CheckerRunning"

893

value_to_set = True

894

895

896

class StopChecker(PropertySetter):

897

propname = "CheckerRunning"

898

value_to_set = False

899

900

901

class ApproveByDefault(PropertySetter):

902

propname = "ApprovedByDefault"

903

value_to_set = True

904

905

906

class DenyByDefault(PropertySetter):

907

propname = "ApprovedByDefault"

908

value_to_set = False

909

910

911

class PropertySetterValue(PropertySetter):

912

"""Abstract class for PropertySetter recieving a value as

913

constructor argument instead of a class attribute."""

767

class StartChecker(Property):

768

propname = "CheckerRunning"

769

value_to_set = dbus.Boolean(True)

770

771

772

class StopChecker(Property):

773

propname = "CheckerRunning"

774

value_to_set = dbus.Boolean(False)

775

776

777

class ApproveByDefault(Property):

778

propname = "ApprovedByDefault"

779

value_to_set = dbus.Boolean(True)

780

781

782

class DenyByDefault(Property):

783

propname = "ApprovedByDefault"

784

value_to_set = dbus.Boolean(False)

785

786

787

class PropertyValue(Property):

788

"Abstract class for Property recieving a value as argument"

914

789

def __init__(self, value):

915

790

self.value_to_set = value

916

791

917

@classmethod

918

def argparse(cls, argtype):

919

def cmdtype(arg):

920

return cls(argtype(arg))

921

return cmdtype

922

792

923

class SetChecker(PropertySetterValue):

793

class SetChecker(PropertyValue):

924

794

propname = "Checker"

925

795

926

796

927

class SetHost(PropertySetterValue):

797

class SetHost(PropertyValue):

928

798

propname = "Host"

929

799

930

800

931

class SetSecret(PropertySetterValue):

801

class SetSecret(PropertyValue):

932

802

propname = "Secret"

933

803

934

804

@property

942

812

value.close()

943

813

944

814

945

class PropertySetterValueMilliseconds(PropertySetterValue):

946

"""Abstract class for PropertySetterValue taking a value

947

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

948

milliseconds."""

815

class MillisecondsPropertyValueArgument(PropertyValue):

816

"""Abstract class for PropertyValue taking a value argument as

817

a datetime.timedelta() but should store it as milliseconds."""

949

818

950

819

@property

951

820

def value_to_set(self):

957

826

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

958

827

959

828

960

class SetTimeout(PropertySetterValueMilliseconds):

829

class SetTimeout(MillisecondsPropertyValueArgument):

961

830

propname = "Timeout"

962

831

963

832

964

class SetExtendedTimeout(PropertySetterValueMilliseconds):

833

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

965

834

propname = "ExtendedTimeout"

966

835

967

836

968

class SetInterval(PropertySetterValueMilliseconds):

837

class SetInterval(MillisecondsPropertyValueArgument):

969

838

propname = "Interval"

970

839

971

840

972

class SetApprovalDelay(PropertySetterValueMilliseconds):

841

class SetApprovalDelay(MillisecondsPropertyValueArgument):

973

842

propname = "ApprovalDelay"

974

843

975

844

976

class SetApprovalDuration(PropertySetterValueMilliseconds):

845

class SetApprovalDuration(MillisecondsPropertyValueArgument):

977

846

propname = "ApprovalDuration"

978

847

979

848

1012

881

"output"))

1013

882

1014

883

1015

class Unique(object):

1016

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

1017

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

1018

1019

1020

884

class Test_string_to_delta(TestCaseWithAssertLogs):

1021

885

# Just test basic RFC 3339 functionality here, the doc string for

1022

886

# rfc3339_duration_to_delta() already has more comprehensive

1023

# tests, which are run by doctest.

887

# tests, which is run by doctest.

1024

888

1025

889

def test_rfc3339_zero_seconds(self):

1026

890

self.assertEqual(datetime.timedelta(),

1050

914

1051

915

def test_actions_requires_client_or_all(self):

1052

916

for action, value in self.actions.items():

1053

args = self.actionargs(action, value)

917

options = self.parser.parse_args()

918

setattr(options, action, value)

1054

919

with self.assertParseError():

1055

self.parse_args(args)

920

self.check_option_syntax(options)

1056

921

1057

# This mostly corresponds to the definition from has_commands() in

922

# This mostly corresponds to the definition from has_actions() in

1058

923

# check_option_syntax()

1059

924

actions = {

1060

"--enable": None,

1061

"--disable": None,

1062

"--bump-timeout": None,

1063

"--start-checker": None,

1064

"--stop-checker": None,

1065

"--is-enabled": None,

1066

"--remove": None,

1067

"--checker": "x",

1068

"--timeout": "PT0S",

1069

"--extended-timeout": "PT0S",

1070

"--interval": "PT0S",

1071

"--approve-by-default": None,

1072

"--deny-by-default": None,

1073

"--approval-delay": "PT0S",

1074

"--approval-duration": "PT0S",

1075

"--host": "hostname",

1076

"--secret": "/dev/null",

1077

"--approve": None,

1078

"--deny": None,

925

# The actual values set here are not that important, but we do

926

# at least stick to the correct types, even though they are

927

# never used

928

"enable": True,

929

"disable": True,

930

"bump_timeout": True,

931

"start_checker": True,

932

"stop_checker": True,

933

"is_enabled": True,

934

"remove": True,

935

"checker": "x",

936

"timeout": datetime.timedelta(),

937

"extended_timeout": datetime.timedelta(),

938

"interval": datetime.timedelta(),

939

"approved_by_default": True,

940

"approval_delay": datetime.timedelta(),

941

"approval_duration": datetime.timedelta(),

942

"host": "x",

943

"secret": io.BytesIO(b"x"),

944

"approve": True,

945

"deny": True,

1079

946

}

1080

947

1081

@staticmethod

1082

def actionargs(action, value, *args):

1083

if value is not None:

1084

return [action, value] + list(args)

1085

else:

1086

return [action] + list(args)

1087

1088

948

@contextlib.contextmanager

1089

949

def assertParseError(self):

1090

950

with self.assertRaises(SystemExit) as e:

1095

955

# /argparse.html#exiting-methods

1096

956

self.assertEqual(2, e.exception.code)

1097

957

1098

def parse_args(self, args):

1099

options = self.parser.parse_args(args)

1100

check_option_syntax(self.parser, options)

1101

1102

958

@staticmethod

1103

959

@contextlib.contextmanager

1104

960

def redirect_stderr_to_devnull():

1115

971

1116

972

def test_actions_all_conflicts_with_verbose(self):

1117

973

for action, value in self.actions.items():

1118

args = self.actionargs(action, value, "--all",

1119

"--verbose")

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.all = True

977

options.verbose = True

1120

978

with self.assertParseError():

1121

self.parse_args(args)

979

self.check_option_syntax(options)

1122

980

1123

981

def test_actions_with_client_conflicts_with_verbose(self):

1124

982

for action, value in self.actions.items():

1125

args = self.actionargs(action, value, "--verbose",

1126

"client")

983

options = self.parser.parse_args()

984

setattr(options, action, value)

985

options.verbose = True

986

options.client = ["client"]

1127

987

with self.assertParseError():

1128

self.parse_args(args)

988

self.check_option_syntax(options)

1129

989

1130

990

def test_dump_json_conflicts_with_verbose(self):

1131

args = ["--dump-json", "--verbose"]

991

options = self.parser.parse_args()

992

options.dump_json = True

993

options.verbose = True

1132

994

with self.assertParseError():

1133

self.parse_args(args)

995

self.check_option_syntax(options)

1134

996

1135

997

def test_dump_json_conflicts_with_action(self):

1136

998

for action, value in self.actions.items():

1137

args = self.actionargs(action, value, "--dump-json")

999

options = self.parser.parse_args()

1000

setattr(options, action, value)

1001

options.dump_json = True

1138

1002

with self.assertParseError():

1139

self.parse_args(args)

1003

self.check_option_syntax(options)

1140

1004

1141

1005

def test_all_can_not_be_alone(self):

1142

args = ["--all"]

1006

options = self.parser.parse_args()

1007

options.all = True

1143

1008

with self.assertParseError():

1144

self.parse_args(args)

1009

self.check_option_syntax(options)

1145

1010

1146

1011

def test_all_is_ok_with_any_action(self):

1147

1012

for action, value in self.actions.items():

1148

args = self.actionargs(action, value, "--all")

1149

self.parse_args(args)

1013

options = self.parser.parse_args()

1014

setattr(options, action, value)

1015

options.all = True

1016

self.check_option_syntax(options)

1150

1017

1151

1018

def test_any_action_is_ok_with_one_client(self):

1152

1019

for action, value in self.actions.items():

1153

args = self.actionargs(action, value, "client")

1154

self.parse_args(args)

1020

options = self.parser.parse_args()

1021

setattr(options, action, value)

1022

options.client = ["client"]

1023

self.check_option_syntax(options)

1155

1024

1156

1025

def test_one_client_with_all_actions_except_is_enabled(self):

1026

options = self.parser.parse_args()

1157

1027

for action, value in self.actions.items():

1158

if action == "--is-enabled":

1028

if action == "is_enabled":

1159

1029

continue

1160

args = self.actionargs(action, value, "client")

1161

self.parse_args(args)

1030

setattr(options, action, value)

1031

options.client = ["client"]

1032

self.check_option_syntax(options)

1162

1033

1163

1034

def test_two_clients_with_all_actions_except_is_enabled(self):

1035

options = self.parser.parse_args()

1164

1036

for action, value in self.actions.items():

1165

if action == "--is-enabled":

1037

if action == "is_enabled":

1166

1038

continue

1167

args = self.actionargs(action, value, "client1",

1168

"client2")

1169

self.parse_args(args)

1039

setattr(options, action, value)

1040

options.client = ["client1", "client2"]

1041

self.check_option_syntax(options)

1170

1042

1171

1043

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1172

1044

for action, value in self.actions.items():

1173

if action == "--is-enabled":

1045

if action == "is_enabled":

1174

1046

continue

1175

args = self.actionargs(action, value, "client1",

1176

"client2")

1177

self.parse_args(args)

1047

options = self.parser.parse_args()

1048

setattr(options, action, value)

1049

options.client = ["client1", "client2"]

1050

self.check_option_syntax(options)

1178

1051

1179

1052

def test_is_enabled_fails_without_client(self):

1180

args = ["--is-enabled"]

1053

options = self.parser.parse_args()

1054

options.is_enabled = True

1181

1055

with self.assertParseError():

1182

self.parse_args(args)

1056

self.check_option_syntax(options)

1183

1057

1184

1058

def test_is_enabled_fails_with_two_clients(self):

1185

args = ["--is-enabled", "client1", "client2"]

1059

options = self.parser.parse_args()

1060

options.is_enabled = True

1061

options.client = ["client1", "client2"]

1186

1062

with self.assertParseError():

1187

self.parse_args(args)

1063

self.check_option_syntax(options)

1188

1064

1189

1065

def test_remove_can_only_be_combined_with_action_deny(self):

1190

1066

for action, value in self.actions.items():

1191

if action in {"--remove", "--deny"}:

1067

if action in {"remove", "deny"}:

1192

1068

continue

1193

args = self.actionargs(action, value, "--all",

1194

"--remove")

1069

options = self.parser.parse_args()

1070

setattr(options, action, value)

1071

options.all = True

1072

options.remove = True

1195

1073

with self.assertParseError():

1196

self.parse_args(args)

1197

1198

1199

class Test_dbus_exceptions(unittest.TestCase):

1200

1201

def test_dbus_ConnectFailed_is_Error(self):

1202

with self.assertRaises(dbus.Error):

1203

raise dbus.ConnectFailed()

1204

1205

1206

class Test_dbus_MandosBus(unittest.TestCase):

1207

1208

class MockMandosBus(dbus.MandosBus):

1209

def __init__(self):

1210

self._name = "se.recompile.Mandos"

1211

self._server_path = "/"

1212

self._server_interface = "se.recompile.Mandos"

1213

self._client_interface = "se.recompile.Mandos.Client"

1214

self.calls = []

1215

self.call_method_return = Unique()

1216

1217

def call_method(self, methodname, busname, objectpath,

1218

interface, *args):

1219

self.calls.append((methodname, busname, objectpath,

1220

interface, args))

1221

return self.call_method_return

1222

1223

def setUp(self):

1224

self.bus = self.MockMandosBus()

1225

1226

def test_set_client_property(self):

1227

self.bus.set_client_property("objectpath", "key", "value")

1228

expected_call = ("Set", self.bus._name, "objectpath",

1229

"org.freedesktop.DBus.Properties",

1230

(self.bus._client_interface, "key", "value"))

1231

self.assertIn(expected_call, self.bus.calls)

1232

1233

def test_call_client_method(self):

1234

ret = self.bus.call_client_method("objectpath", "methodname")

1235

self.assertIs(self.bus.call_method_return, ret)

1236

expected_call = ("methodname", self.bus._name, "objectpath",

1237

self.bus._client_interface, ())

1238

self.assertIn(expected_call, self.bus.calls)

1239

1240

def test_call_client_method_with_args(self):

1241

args = (Unique(), Unique())

1242

ret = self.bus.call_client_method("objectpath", "methodname",

1243

*args)

1244

self.assertIs(self.bus.call_method_return, ret)

1245

expected_call = ("methodname", self.bus._name, "objectpath",

1246

self.bus._client_interface,

1247

(args[0], args[1]))

1248

self.assertIn(expected_call, self.bus.calls)

1249

1250

def test_get_clients_and_properties(self):

1251

managed_objects = {

1252

"objectpath": {

1253

self.bus._client_interface: {

1254

"key": "value",

1255

"bool": True,

1256

1257

"irrelevant_interface": {

1258

"key": "othervalue",

1259

"bool": False,

1260

1261

1262

"other_objectpath": {

1263

"other_irrelevant_interface": {

1264

"key": "value 3",

1265

"bool": None,

1266

1267

1268

}

1269

expected_clients_and_properties = {

1270

"objectpath": {

1271

"key": "value",

1272

"bool": True,

1273

}

1274

}

1275

self.bus.call_method_return = managed_objects

1276

ret = self.bus.get_clients_and_properties()

1277

self.assertDictEqual(expected_clients_and_properties, ret)

1278

expected_call = ("GetManagedObjects", self.bus._name,

1279

self.bus._server_path,

1280

"org.freedesktop.DBus.ObjectManager", ())

1281

self.assertIn(expected_call, self.bus.calls)

1282

1283

def test_call_server_method(self):

1284

ret = self.bus.call_server_method("methodname")

1285

self.assertIs(self.bus.call_method_return, ret)

1286

expected_call = ("methodname", self.bus._name,

1287

self.bus._server_path,

1288

self.bus._server_interface, ())

1289

self.assertIn(expected_call, self.bus.calls)

1290

1291

def test_call_server_method_with_args(self):

1292

args = (Unique(), Unique())

1293

ret = self.bus.call_server_method("methodname", *args)

1294

self.assertIs(self.bus.call_method_return, ret)

1295

expected_call = ("methodname", self.bus._name,

1296

self.bus._server_path,

1297

self.bus._server_interface,

1298

(args[0], args[1]))

1299

self.assertIn(expected_call, self.bus.calls)

1300

1301

1302

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1303

1304

def MockDBusPython_func(self, func):

1305

class mock_dbus_python(object):

1306

"""mock dbus-python module"""

1307

class exceptions(object):

1308

"""Pseudo-namespace"""

1309

class DBusException(Exception):

1310

pass

1311

class SystemBus(object):

1312

@staticmethod

1313

def get_object(busname, objectpath):

1314

DBusObject = collections.namedtuple(

1315

"DBusObject", ("methodname",))

1316

def method(*args, **kwargs):

1317

self.assertEqual({"dbus_interface":

1318

"interface"},

1319

kwargs)

1320

return func(*args)

1321

return DBusObject(methodname=method)

1322

class Boolean(object):

1323

def __init__(self, value):

1324

self.value = bool(value)

1325

def __bool__(self):

1326

return self.value

1327

if sys.version_info.major == 2:

1328

__nonzero__ = __bool__

1329

class ObjectPath(str):

1330

pass

1331

class Dictionary(dict):

1332

pass

1333

return mock_dbus_python

1334

1335

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

1336

interface, *args):

1337

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

1338

return bus.call_method(methodname, busname, objectpath,

1339

interface, *args)

1340

1341

def test_call_method_returns(self):

1342

expected_method_return = Unique()

1343

method_args = (Unique(), Unique())

1344

def func(*args):

1345

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

1346

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

1347

self.assertIs(marg, arg)

1348

return expected_method_return

1349

mock_dbus_python = self.MockDBusPython_func(func)

1350

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1351

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

1352

"objectpath", "interface",

1353

*method_args)

1354

self.assertIs(ret, expected_method_return)

1355

1356

def test_call_method_filters_bool_true(self):

1357

def func():

1358

return method_return

1359

mock_dbus_python = self.MockDBusPython_func(func)

1360

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1361

method_return = mock_dbus_python.Boolean(True)

1362

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

1363

"objectpath", "interface")

1364

self.assertTrue(ret)

1365

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1366

1367

def test_call_method_filters_bool_false(self):

1368

def func():

1369

return method_return

1370

mock_dbus_python = self.MockDBusPython_func(func)

1371

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1372

method_return = mock_dbus_python.Boolean(False)

1373

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

1374

"objectpath", "interface")

1375

self.assertFalse(ret)

1376

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1377

1378

def test_call_method_filters_objectpath(self):

1379

def func():

1380

return method_return

1381

mock_dbus_python = self.MockDBusPython_func(func)

1382

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1383

method_return = mock_dbus_python.ObjectPath("objectpath")

1384

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

1385

"objectpath", "interface")

1386

self.assertEqual("objectpath", ret)

1387

self.assertIsNot("objectpath", ret)

1388

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1389

1390

def test_call_method_filters_booleans_in_dict(self):

1391

def func():

1392

return method_return

1393

mock_dbus_python = self.MockDBusPython_func(func)

1394

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1395

method_return = mock_dbus_python.Dictionary(

1396

{mock_dbus_python.Boolean(True):

1397

mock_dbus_python.Boolean(False),

1398

mock_dbus_python.Boolean(False):

1399

mock_dbus_python.Boolean(True)})

1400

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

1401

"objectpath", "interface")

1402

expected_method_return = {True: False,

1403

False: True}

1404

self.assertEqual(expected_method_return, ret)

1405

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1406

1407

def test_call_method_filters_objectpaths_in_dict(self):

1408

def func():

1409

return method_return

1410

mock_dbus_python = self.MockDBusPython_func(func)

1411

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1412

method_return = mock_dbus_python.Dictionary(

1413

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1414

mock_dbus_python.ObjectPath("objectpath_value_1"),

1415

mock_dbus_python.ObjectPath("objectpath_key_2"):

1416

mock_dbus_python.ObjectPath("objectpath_value_2")})

1417

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

1418

"objectpath", "interface")

1419

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

1420

for key, value in

1421

method_return.items()}

1422

self.assertEqual(expected_method_return, ret)

1423

self.assertIsInstance(ret, dict)

1424

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1425

1426

def test_call_method_filters_dict_in_dict(self):

1427

def func():

1428

return method_return

1429

mock_dbus_python = self.MockDBusPython_func(func)

1430

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1431

method_return = mock_dbus_python.Dictionary(

1432

{"key1": mock_dbus_python.Dictionary({"key11": "value11",

1433

"key12": "value12"}),

1434

"key2": mock_dbus_python.Dictionary({"key21": "value21",

1435

"key22": "value22"})})

1436

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

1437

"objectpath", "interface")

1438

expected_method_return = {

1439

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

1440

"key12": "value12"},

1441

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

1442

"key22": "value22"},

1443

}

1444

self.assertEqual(expected_method_return, ret)

1445

self.assertIsInstance(ret, dict)

1446

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1447

for key, value in ret.items():

1448

self.assertIsInstance(value, dict)

1449

self.assertEqual(expected_method_return[key], value)

1450

self.assertNotIsInstance(value,

1451

mock_dbus_python.Dictionary)

1452

1453

def test_call_method_filters_dict_three_deep(self):

1454

def func():

1455

return method_return

1456

mock_dbus_python = self.MockDBusPython_func(func)

1457

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1458

method_return = mock_dbus_python.Dictionary(

1459

{"key1":

1460

mock_dbus_python.Dictionary(

1461

{"key2":

1462

mock_dbus_python.Dictionary(

1463

{"key3":

1464

mock_dbus_python.Boolean(True),

1465

}),

1466

}),

1467

})

1468

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

1469

"objectpath", "interface")

1470

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

1471

self.assertEqual(expected_method_return, ret)

1472

self.assertIsInstance(ret, dict)

1473

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1474

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

1475

self.assertNotIsInstance(ret["key1"],

1476

mock_dbus_python.Dictionary)

1477

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

1478

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

1479

mock_dbus_python.Dictionary)

1480

self.assertTrue(ret["key1"]["key2"]["key3"])

1481

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

1482

mock_dbus_python.Boolean)

1483

1484

def test_call_method_handles_exception(self):

1074

self.check_option_syntax(options)

1075

1076

1077

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1078

def test_calls_and_returns_get_object_on_bus(self):

1079

class MockBus(object):

1080

called = False

1081

def get_object(mockbus_self, busname, dbus_path):

1082

# Note that "self" is still the testcase instance,

1083

# this MockBus instance is in "mockbus_self".

1084

self.assertEqual(dbus_busname, busname)

1085

self.assertEqual(server_dbus_path, dbus_path)

1086

mockbus_self.called = True

1087

return mockbus_self

1088

1089

mockbus = get_mandos_dbus_object(bus=MockBus())

1090

self.assertIsInstance(mockbus, MockBus)

1091

self.assertTrue(mockbus.called)

1092

1093

def test_logs_and_exits_on_dbus_error(self):

1094

class FailingBusStub(object):

1095

def get_object(self, busname, dbus_path):

1096

raise dbus.exceptions.DBusException("Test")

1097

1098

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

1099

with self.assertRaises(SystemExit) as e:

1100

bus = get_mandos_dbus_object(bus=FailingBusStub())

1101

1102

if isinstance(e.exception.code, int):

1103

self.assertNotEqual(0, e.exception.code)

1104

else:

1105

self.assertIsNotNone(e.exception.code)

1106

1107

1108

class Test_get_managed_objects(TestCaseWithAssertLogs):

1109

def test_calls_and_returns_GetManagedObjects(self):

1110

managed_objects = {"/clients/client": { "Name": "client"}}

1111

class ObjectManagerStub(object):

1112

def GetManagedObjects(self):

1113

return managed_objects

1114

retval = get_managed_objects(ObjectManagerStub())

1115

self.assertDictEqual(managed_objects, retval)

1116

1117

def test_logs_and_exits_on_dbus_error(self):

1485

1118

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

1486

1119

1487

def func():

1488

dbus_logger.error("Test")

1489

raise mock_dbus_python.exceptions.DBusException()

1490

1491

mock_dbus_python = self.MockDBusPython_func(func)

1492

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1120

class ObjectManagerFailingStub(object):

1121

def GetManagedObjects(self):

1122

dbus_logger.error("Test")

1123

raise dbus.exceptions.DBusException("Test")

1493

1124

1494

1125

class CountingHandler(logging.Handler):

1495

1126

count = 0

1501

1132

dbus_logger.addHandler(counting_handler)

1502

1133

1503

1134

try:

1504

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

1505

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

1506

"objectpath", "interface")

1135

with self.assertLogs(log, logging.CRITICAL) as watcher:

1136

with self.assertRaises(SystemExit) as e:

1137

get_managed_objects(ObjectManagerFailingStub())

1507

1138

finally:

1508

1139

dbus_logger.removeFilter(counting_handler)

1509

1140

1510

self.assertNotIsInstance(e, dbus.ConnectFailed)

1511

1512

1141

# Make sure the dbus logger was suppressed

1513

1142

self.assertEqual(0, counting_handler.count)

1514

1143

1515

def test_get_object_converts_to_correct_exception(self):

1516

bus = dbus_python_adapter.SystemBus(

1517

self.fake_dbus_python_raises_exception_on_connect)

1518

with self.assertRaises(dbus.ConnectFailed):

1519

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

1520

"objectpath", "interface")

1521

1522

class fake_dbus_python_raises_exception_on_connect(object):

1523

"""fake dbus-python module"""

1524

class exceptions(object):

1525

"""Pseudo-namespace"""

1526

class DBusException(Exception):

1527

pass

1528

1529

@classmethod

1530

def SystemBus(cls):

1531

def get_object(busname, objectpath):

1532

raise cls.exceptions.DBusException()

1533

Bus = collections.namedtuple("Bus", ["get_object"])

1534

return Bus(get_object=get_object)

1535

1536

1537

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1538

class mock_dbus_python(object):

1539

"""mock dbus-python modules"""

1540

class SystemBus(object):

1541

@staticmethod

1542

def get_object(busname, objectpath):

1543

return Unique()

1544

1545

def setUp(self):

1546

self.bus = dbus_python_adapter.CachingBus(

1547

self.mock_dbus_python)

1548

1549

def test_returns_distinct_objectpaths(self):

1550

obj1 = self.bus.get_object("busname", "objectpath1")

1551

self.assertIsInstance(obj1, Unique)

1552

obj2 = self.bus.get_object("busname", "objectpath2")

1553

self.assertIsInstance(obj2, Unique)

1554

self.assertIsNot(obj1, obj2)

1555

1556

def test_returns_distinct_busnames(self):

1557

obj1 = self.bus.get_object("busname1", "objectpath")

1558

self.assertIsInstance(obj1, Unique)

1559

obj2 = self.bus.get_object("busname2", "objectpath")

1560

self.assertIsInstance(obj2, Unique)

1561

self.assertIsNot(obj1, obj2)

1562

1563

def test_returns_distinct_both(self):

1564

obj1 = self.bus.get_object("busname1", "objectpath")

1565

self.assertIsInstance(obj1, Unique)

1566

obj2 = self.bus.get_object("busname2", "objectpath")

1567

self.assertIsInstance(obj2, Unique)

1568

self.assertIsNot(obj1, obj2)

1569

1570

def test_returns_same(self):

1571

obj1 = self.bus.get_object("busname", "objectpath")

1572

self.assertIsInstance(obj1, Unique)

1573

obj2 = self.bus.get_object("busname", "objectpath")

1574

self.assertIsInstance(obj2, Unique)

1575

self.assertIs(obj1, obj2)

1576

1577

def test_returns_same_old(self):

1578

obj1 = self.bus.get_object("busname1", "objectpath1")

1579

self.assertIsInstance(obj1, Unique)

1580

obj2 = self.bus.get_object("busname2", "objectpath2")

1581

self.assertIsInstance(obj2, Unique)

1582

obj1b = self.bus.get_object("busname1", "objectpath1")

1583

self.assertIsInstance(obj1b, Unique)

1584

self.assertIsNot(obj1, obj2)

1585

self.assertIsNot(obj2, obj1b)

1586

self.assertIs(obj1, obj1b)

1587

1588

1589

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1590

1591

def Stub_pydbus_func(self, func):

1592

class stub_pydbus(object):

1593

"""stub pydbus module"""

1594

class SystemBus(object):

1595

@staticmethod

1596

def get(busname, objectpath):

1597

DBusObject = collections.namedtuple(

1598

"DBusObject", ("methodname",))

1599

return {"interface":

1600

DBusObject(methodname=func)}

1601

return stub_pydbus

1602

1603

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

1604

interface, *args):

1605

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

1606

return bus.call_method(methodname, busname, objectpath,

1607

interface, *args)

1608

1609

def test_call_method_returns(self):

1610

expected_method_return = Unique()

1611

method_args = (Unique(), Unique())

1612

def func(*args):

1613

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

1614

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

1615

self.assertIs(marg, arg)

1616

return expected_method_return

1617

stub_pydbus = self.Stub_pydbus_func(func)

1618

bus = pydbus_adapter.SystemBus(stub_pydbus)

1619

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

1620

"objectpath", "interface",

1621

*method_args)

1622

self.assertIs(ret, expected_method_return)

1623

1624

def test_call_method_handles_exception(self):

1625

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

1626

1627

def func():

1628

raise gi.repository.GLib.Error()

1629

1630

stub_pydbus = self.Stub_pydbus_func(func)

1631

bus = pydbus_adapter.SystemBus(stub_pydbus)

1632

1633

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

1634

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

1635

"objectpath", "interface")

1636

1637

self.assertNotIsInstance(e, dbus.ConnectFailed)

1638

1639

def test_get_converts_to_correct_exception(self):

1640

bus = pydbus_adapter.SystemBus(

1641

self.fake_pydbus_raises_exception_on_connect)

1642

with self.assertRaises(dbus.ConnectFailed):

1643

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

1644

"objectpath", "interface")

1645

1646

class fake_pydbus_raises_exception_on_connect(object):

1647

"""fake dbus-python module"""

1648

@classmethod

1649

def SystemBus(cls):

1650

def get(busname, objectpath):

1651

raise gi.repository.GLib.Error()

1652

Bus = collections.namedtuple("Bus", ["get"])

1653

return Bus(get=get)

1654

1655

def test_set_property_uses_setattr(self):

1656

class Object(object):

1657

pass

1658

obj = Object()

1659

class pydbus_spy(object):

1660

class SystemBus(object):

1661

@staticmethod

1662

def get(busname, objectpath):

1663

return {"interface": obj}

1664

bus = pydbus_adapter.SystemBus(pydbus_spy)

1665

value = Unique()

1666

bus.set_property("busname", "objectpath", "interface", "key",

1667

value)

1668

self.assertIs(value, obj.key)

1669

1670

def test_get_suppresses_xml_deprecation_warning(self):

1671

if sys.version_info.major >= 3:

1672

return

1673

class stub_pydbus_get(object):

1674

class SystemBus(object):

1675

@staticmethod

1676

def get(busname, objectpath):

1677

warnings.warn_explicit(

1678

"deprecated", DeprecationWarning,

1679

"xml.etree.ElementTree", 0)

1680

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1681

with warnings.catch_warnings(record=True) as w:

1682

warnings.simplefilter("always")

1683

bus.get("busname", "objectpath")

1684

self.assertEqual(0, len(w))

1685

1686

1687

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1688

class stub_pydbus(object):

1689

"""stub pydbus module"""

1690

class SystemBus(object):

1691

@staticmethod

1692

def get(busname, objectpath):

1693

return Unique()

1694

1695

def setUp(self):

1696

self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)

1697

1698

def test_returns_distinct_objectpaths(self):

1699

obj1 = self.bus.get("busname", "objectpath1")

1700

self.assertIsInstance(obj1, Unique)

1701

obj2 = self.bus.get("busname", "objectpath2")

1702

self.assertIsInstance(obj2, Unique)

1703

self.assertIsNot(obj1, obj2)

1704

1705

def test_returns_distinct_busnames(self):

1706

obj1 = self.bus.get("busname1", "objectpath")

1707

self.assertIsInstance(obj1, Unique)

1708

obj2 = self.bus.get("busname2", "objectpath")

1709

self.assertIsInstance(obj2, Unique)

1710

self.assertIsNot(obj1, obj2)

1711

1712

def test_returns_distinct_both(self):

1713

obj1 = self.bus.get("busname1", "objectpath")

1714

self.assertIsInstance(obj1, Unique)

1715

obj2 = self.bus.get("busname2", "objectpath")

1716

self.assertIsInstance(obj2, Unique)

1717

self.assertIsNot(obj1, obj2)

1718

1719

def test_returns_same(self):

1720

obj1 = self.bus.get("busname", "objectpath")

1721

self.assertIsInstance(obj1, Unique)

1722

obj2 = self.bus.get("busname", "objectpath")

1723

self.assertIsInstance(obj2, Unique)

1724

self.assertIs(obj1, obj2)

1725

1726

def test_returns_same_old(self):

1727

obj1 = self.bus.get("busname1", "objectpath1")

1728

self.assertIsInstance(obj1, Unique)

1729

obj2 = self.bus.get("busname2", "objectpath2")

1730

self.assertIsInstance(obj2, Unique)

1731

obj1b = self.bus.get("busname1", "objectpath1")

1732

self.assertIsInstance(obj1b, Unique)

1733

self.assertIsNot(obj1, obj2)

1734

self.assertIsNot(obj2, obj1b)

1735

self.assertIs(obj1, obj1b)

1144

# Test that the dbus_logger still works

1145

with self.assertLogs(dbus_logger, logging.ERROR):

1146

dbus_logger.error("Test")

1147

1148

if isinstance(e.exception.code, int):

1149

self.assertNotEqual(0, e.exception.code)

1150

else:

1151

self.assertIsNotNone(e.exception.code)

1736

1152

1737

1153

1738

1154

class Test_commands_from_options(unittest.TestCase):

1739

1740

1155

def setUp(self):

1741

1156

self.parser = argparse.ArgumentParser()

1742

1157

add_command_line_options(self.parser)

1957

1372

class TestCommand(unittest.TestCase):

1958

1373

"""Abstract class for tests of command classes"""

1959

1374

1960

class FakeMandosBus(dbus.MandosBus):

1961

def __init__(self, testcase):

1962

self.client_properties = {

1963

"Name": "foo",

1964

"KeyID": ("92ed150794387c03ce684574b1139a65"

1965

"94a34f895daaaf09fd8ea90a27cddb12"),

1966

"Secret": b"secret",

1967

"Host": "foo.example.org",

1968

"Enabled": True,

1969

"Timeout": 300000,

1970

"LastCheckedOK": "2019-02-03T00:00:00",

1971

"Created": "2019-01-02T00:00:00",

1972

"Interval": 120000,

1973

"Fingerprint": ("778827225BA7DE539C5A"

1974

"7CFA59CFF7CDBD9A5920"),

1975

"CheckerRunning": False,

1976

"LastEnabled": "2019-01-03T00:00:00",

1977

"ApprovalPending": False,

1978

"ApprovedByDefault": True,

1979

"LastApprovalRequest": "",

1980

"ApprovalDelay": 0,

1981

"ApprovalDuration": 1000,

1982

"Checker": "fping -q -- %(host)s",

1983

"ExtendedTimeout": 900000,

1984

"Expires": "2019-02-04T00:00:00",

1985

"LastCheckerStatus": 0,

1986

}

1987

self.other_client_properties = {

1988

"Name": "barbar",

1989

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1990

"6ab612cff5ad227247e46c2b020f441c"),

1991

"Secret": b"secretbar",

1992

"Host": "192.0.2.3",

1993

"Enabled": True,

1994

"Timeout": 300000,

1995

"LastCheckedOK": "2019-02-04T00:00:00",

1996

"Created": "2019-01-03T00:00:00",

1997

"Interval": 120000,

1998

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1999

"F547B3A107558FCA3A27"),

2000

"CheckerRunning": True,

2001

"LastEnabled": "2019-01-04T00:00:00",

2002

"ApprovalPending": False,

2003

"ApprovedByDefault": False,

2004

"LastApprovalRequest": "2019-01-03T00:00:00",

2005

"ApprovalDelay": 30000,

2006

"ApprovalDuration": 93785000,

2007

"Checker": ":",

2008

"ExtendedTimeout": 900000,

2009

"Expires": "2019-02-05T00:00:00",

2010

"LastCheckerStatus": -2,

2011

}

2012

self.clients = collections.OrderedDict(

2013

[

2014

("client_objectpath", self.client_properties),

2015

("other_client_objectpath",

2016

self.other_client_properties),

2017

])

2018

self.one_client = {"client_objectpath":

2019

self.client_properties}

2020

self.testcase = testcase

2021

self.calls = []

2022

2023

def call_method(self, methodname, busname, objectpath,

2024

interface, *args):

2025

self.testcase.assertEqual("se.recompile.Mandos", busname)

2026

self.calls.append((methodname, busname, objectpath,

2027

interface, args))

2028

if interface == "org.freedesktop.DBus.Properties":

2029

if methodname == "Set":

2030

self.testcase.assertEqual(3, len(args))

2031

interface, key, value = args

2032

self.testcase.assertEqual(

2033

"se.recompile.Mandos.Client", interface)

2034

self.clients[objectpath][key] = value

2035

return

2036

elif interface == "se.recompile.Mandos":

2037

self.testcase.assertEqual("RemoveClient", methodname)

2038

self.testcase.assertEqual(1, len(args))

2039

clientpath = args[0]

2040

del self.clients[clientpath]

2041

return

2042

elif interface == "se.recompile.Mandos.Client":

2043

if methodname == "Approve":

2044

self.testcase.assertEqual(1, len(args))

2045

return

2046

raise ValueError()

2047

2048

1375

def setUp(self):

2049

self.bus = self.FakeMandosBus(self)

1376

testcase = self

1377

class MockClient(object):

1378

def __init__(self, name, **attributes):

1379

self.__dbus_object_path__ = "/clients/{}".format(name)

1380

self.attributes = attributes

1381

self.attributes["Name"] = name

1382

self.calls = []

1383

def Set(self, interface, propname, value, dbus_interface):

1384

testcase.assertEqual(client_dbus_interface, interface)

1385

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1386

dbus_interface)

1387

self.attributes[propname] = value

1388

def Approve(self, approve, dbus_interface):

1389

testcase.assertEqual(client_dbus_interface,

1390

dbus_interface)

1391

self.calls.append(("Approve", (approve,

1392

dbus_interface)))

1393

self.client = MockClient(

1394

"foo",

1395

KeyID=("92ed150794387c03ce684574b1139a65"

1396

"94a34f895daaaf09fd8ea90a27cddb12"),

1397

Secret=b"secret",

1398

Host="foo.example.org",

1399

Enabled=dbus.Boolean(True),

1400

Timeout=300000,

1401

LastCheckedOK="2019-02-03T00:00:00",

1402

Created="2019-01-02T00:00:00",

1403

Interval=120000,

1404

Fingerprint=("778827225BA7DE539C5A"

1405

"7CFA59CFF7CDBD9A5920"),

1406

CheckerRunning=dbus.Boolean(False),

1407

LastEnabled="2019-01-03T00:00:00",

1408

ApprovalPending=dbus.Boolean(False),

1409

ApprovedByDefault=dbus.Boolean(True),

1410

LastApprovalRequest="",

1411

ApprovalDelay=0,

1412

ApprovalDuration=1000,

1413

Checker="fping -q -- %(host)s",

1414

ExtendedTimeout=900000,

1415

Expires="2019-02-04T00:00:00",

1416

LastCheckerStatus=0)

1417

self.other_client = MockClient(

1418

"barbar",

1419

KeyID=("0558568eedd67d622f5c83b35a115f79"

1420

"6ab612cff5ad227247e46c2b020f441c"),

1421

Secret=b"secretbar",

1422

Host="192.0.2.3",

1423

Enabled=dbus.Boolean(True),

1424

Timeout=300000,

1425

LastCheckedOK="2019-02-04T00:00:00",

1426

Created="2019-01-03T00:00:00",

1427

Interval=120000,

1428

Fingerprint=("3E393AEAEFB84C7E89E2"

1429

"F547B3A107558FCA3A27"),

1430

CheckerRunning=dbus.Boolean(True),

1431

LastEnabled="2019-01-04T00:00:00",

1432

ApprovalPending=dbus.Boolean(False),

1433

ApprovedByDefault=dbus.Boolean(False),

1434

LastApprovalRequest="2019-01-03T00:00:00",

1435

ApprovalDelay=30000,

1436

ApprovalDuration=93785000,

1437

Checker=":",

1438

ExtendedTimeout=900000,

1439

Expires="2019-02-05T00:00:00",

1440

LastCheckerStatus=-2)

1441

self.clients = collections.OrderedDict(

1442

[

1443

(self.client.__dbus_object_path__,

1444

self.client.attributes),

1445

(self.other_client.__dbus_object_path__,

1446

self.other_client.attributes),

1447

])

1448

self.one_client = {self.client.__dbus_object_path__:

1449

self.client.attributes}

1450

1451

@property

1452

def bus(self):

1453

class MockBus(object):

1454

@staticmethod

1455

def get_object(client_bus_name, path):

1456

self.assertEqual(dbus_busname, client_bus_name)

1457

# Note: "self" here is the TestCmd instance, not the

1458

# MockBus instance, since this is a static method!

1459

if path == self.client.__dbus_object_path__:

1460

return self.client

1461

elif path == self.other_client.__dbus_object_path__:

1462

return self.other_client

1463

return MockBus()

2050

1464

2051

1465

2052

1466

class TestBaseCommands(TestCommand):

2053

1467

2054

1468

def test_IsEnabled_exits_successfully(self):

2055

1469

with self.assertRaises(SystemExit) as e:

2056

command.IsEnabled().run(self.bus.one_client)

1470

command.IsEnabled().run(self.one_client)

2057

1471

if e.exception.code is not None:

2058

1472

self.assertEqual(0, e.exception.code)

2059

1473

else:

2060

1474

self.assertIsNone(e.exception.code)

2061

1475

2062

1476

def test_IsEnabled_exits_with_failure(self):

2063

self.bus.client_properties["Enabled"] = False

1477

self.client.attributes["Enabled"] = dbus.Boolean(False)

2064

1478

with self.assertRaises(SystemExit) as e:

2065

command.IsEnabled().run(self.bus.one_client)

1479

command.IsEnabled().run(self.one_client)

2066

1480

if isinstance(e.exception.code, int):

2067

1481

self.assertNotEqual(0, e.exception.code)

2068

1482

else:

2069

1483

self.assertIsNotNone(e.exception.code)

2070

1484

2071

1485

def test_Approve(self):

2072

busname = "se.recompile.Mandos"

2073

client_interface = "se.recompile.Mandos.Client"

2074

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

2075

for clientpath in self.bus.clients:

2076

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

2077

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

1486

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

1487

for clientpath in self.clients:

1488

client = self.bus.get_object(dbus_busname, clientpath)

1489

self.assertIn(("Approve", (True, client_dbus_interface)),

1490

client.calls)

2078

1491

2079

1492

def test_Deny(self):

2080

busname = "se.recompile.Mandos"

2081

client_interface = "se.recompile.Mandos.Client"

2082

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

2083

for clientpath in self.bus.clients:

2084

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

2085

client_interface, (False,)),

2086

self.bus.calls)

1493

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

1494

for clientpath in self.clients:

1495

client = self.bus.get_object(dbus_busname, clientpath)

1496

self.assertIn(("Approve", (False, client_dbus_interface)),

1497

client.calls)

2087

1498

2088

1499

def test_Remove(self):

2089

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,

2093

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

1500

class MandosSpy(object):

1501

def __init__(self):

1502

self.calls = []

1503

def RemoveClient(self, dbus_path):

1504

self.calls.append(("RemoveClient", (dbus_path,)))

1505

mandos = MandosSpy()

1506

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

1507

for clientpath in self.clients:

1508

self.assertIn(("RemoveClient", (clientpath,)),

1509

mandos.calls)

2094

1510

2095

1511

expected_json = {

2096

1512

"foo": {

2145

1561

2146

1562

def test_DumpJSON_normal(self):

2147

1563

with self.capture_stdout_to_buffer() as buffer:

2148

command.DumpJSON().run(self.bus.clients)

1564

command.DumpJSON().run(self.clients)

2149

1565

json_data = json.loads(buffer.getvalue())

2150

1566

self.assertDictEqual(self.expected_json, json_data)

2151

1567

2162

1578

2163

1579

def test_DumpJSON_one_client(self):

2164

1580

with self.capture_stdout_to_buffer() as buffer:

2165

command.DumpJSON().run(self.bus.one_client)

1581

command.DumpJSON().run(self.one_client)

2166

1582

json_data = json.loads(buffer.getvalue())

2167

1583

expected_json = {"foo": self.expected_json["foo"]}

2168

1584

self.assertDictEqual(expected_json, json_data)

2169

1585

2170

1586

def test_PrintTable_normal(self):

2171

1587

with self.capture_stdout_to_buffer() as buffer:

2172

command.PrintTable().run(self.bus.clients)

1588

command.PrintTable().run(self.clients)

2173

1589

expected_output = "\n".join((

2174

1590

"Name Enabled Timeout Last Successful Check",

2175

1591

"foo Yes 00:05:00 2019-02-03T00:00:00 ",

2179

1595

2180

1596

def test_PrintTable_verbose(self):

2181

1597

with self.capture_stdout_to_buffer() as buffer:

2182

command.PrintTable(verbose=True).run(self.bus.clients)

1598

command.PrintTable(verbose=True).run(self.clients)

2183

1599

columns = (

2184

1600

(

2185

1601

"Name ",

2275

1691

2276

1692

def test_PrintTable_one_client(self):

2277

1693

with self.capture_stdout_to_buffer() as buffer:

2278

command.PrintTable().run(self.bus.one_client)

1694

command.PrintTable().run(self.one_client)

2279

1695

expected_output = "\n".join((

2280

1696

"Name Enabled Timeout Last Successful Check",

2281

1697

"foo Yes 00:05:00 2019-02-03T00:00:00 ",

2283

1699

self.assertEqual(expected_output, buffer.getvalue())

2284

1700

2285

1701

2286

class TestPropertySetterCmd(TestCommand):

2287

"""Abstract class for tests of command.PropertySetter classes"""

2288

1702

class TestPropertyCmd(TestCommand):

1703

"""Abstract class for tests of command.Property classes"""

2289

1704

def runTest(self):

2290

1705

if not hasattr(self, "command"):

2291

return # Abstract TestCase class

2292

2293

if hasattr(self, "values_to_set"):

2294

cmd_args = [(value,) for value in self.values_to_set]

2295

values_to_get = getattr(self, "values_to_get",

2296

self.values_to_set)

2297

else:

2298

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

2299

values_to_get = self.values_to_get

2300

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

2301

for clientpath in self.bus.clients:

2302

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

2303

Unique())

2304

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

2305

for clientpath in self.bus.clients:

2306

value = (self.bus.clients[clientpath]

2307

[self.propname])

2308

self.assertNotIsInstance(value, Unique)

1706

return

1707

values_to_get = getattr(self, "values_to_get",

1708

self.values_to_set)

1709

for value_to_set, value_to_get in zip(self.values_to_set,

1710

values_to_get):

1711

for clientpath in self.clients:

1712

client = self.bus.get_object(dbus_busname, clientpath)

1713

old_value = client.attributes[self.propname]

1714

client.attributes[self.propname] = self.Unique()

1715

self.run_command(value_to_set, self.clients)

1716

for clientpath in self.clients:

1717

client = self.bus.get_object(dbus_busname, clientpath)

1718

value = client.attributes[self.propname]

1719

self.assertNotIsInstance(value, self.Unique)

2309

1720

self.assertEqual(value_to_get, value)

2310

1721

2311

2312

class TestEnableCmd(TestPropertySetterCmd):

1722

class Unique(object):

1723

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

1724

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

1725

1726

def run_command(self, value, clients):

1727

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

1728

1729

1730

class TestEnableCmd(TestPropertyCmd):

2313

1731

command = command.Enable

2314

1732

propname = "Enabled"

2315

values_to_get = [True]

2316

2317

2318

class TestDisableCmd(TestPropertySetterCmd):

1733

values_to_set = [dbus.Boolean(True)]

1734

1735

1736

class TestDisableCmd(TestPropertyCmd):

2319

1737

command = command.Disable

2320

1738

propname = "Enabled"

2321

values_to_get = [False]

2322

2323

2324

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1739

values_to_set = [dbus.Boolean(False)]

1740

1741

1742

class TestBumpTimeoutCmd(TestPropertyCmd):

2325

1743

command = command.BumpTimeout

2326

1744

propname = "LastCheckedOK"

2327

values_to_get = [""]

2328

2329

2330

class TestStartCheckerCmd(TestPropertySetterCmd):

1745

values_to_set = [""]

1746

1747

1748

class TestStartCheckerCmd(TestPropertyCmd):

2331

1749

command = command.StartChecker

2332

1750

propname = "CheckerRunning"

2333

values_to_get = [True]

2334

2335

2336

class TestStopCheckerCmd(TestPropertySetterCmd):

1751

values_to_set = [dbus.Boolean(True)]

1752

1753

1754

class TestStopCheckerCmd(TestPropertyCmd):

2337

1755

command = command.StopChecker

2338

1756

propname = "CheckerRunning"

2339

values_to_get = [False]

2340

2341

2342

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1757

values_to_set = [dbus.Boolean(False)]

1758

1759

1760

class TestApproveByDefaultCmd(TestPropertyCmd):

2343

1761

command = command.ApproveByDefault

2344

1762

propname = "ApprovedByDefault"

2345

values_to_get = [True]

2346

2347

2348

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1763

values_to_set = [dbus.Boolean(True)]

1764

1765

1766

class TestDenyByDefaultCmd(TestPropertyCmd):

2349

1767

command = command.DenyByDefault

2350

1768

propname = "ApprovedByDefault"

2351

values_to_get = [False]

2352

2353

2354

class TestSetCheckerCmd(TestPropertySetterCmd):

1769

values_to_set = [dbus.Boolean(False)]

1770

1771

1772

class TestPropertyValueCmd(TestPropertyCmd):

1773

"""Abstract class for tests of PropertyValueCmd classes"""

1774

1775

def runTest(self):

1776

if type(self) is TestPropertyValueCmd:

1777

return

1778

return super(TestPropertyValueCmd, self).runTest()

1779

1780

def run_command(self, value, clients):

1781

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

1782

1783

1784

class TestSetCheckerCmd(TestPropertyValueCmd):

2355

1785

command = command.SetChecker

2356

1786

propname = "Checker"

2357

1787

values_to_set = ["", ":", "fping -q -- %s"]

2358

1788

2359

1789

2360

class TestSetHostCmd(TestPropertySetterCmd):

1790

class TestSetHostCmd(TestPropertyValueCmd):

2361

1791

command = command.SetHost

2362

1792

propname = "Host"

2363

1793

values_to_set = ["192.0.2.3", "client.example.org"]

2364

1794

2365

1795

2366

class TestSetSecretCmd(TestPropertySetterCmd):

1796

class TestSetSecretCmd(TestPropertyValueCmd):

2367

1797

command = command.SetSecret

2368

1798

propname = "Secret"

2369

1799

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

2371

1801

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

2372

1802

2373

1803

2374

class TestSetTimeoutCmd(TestPropertySetterCmd):

1804

class TestSetTimeoutCmd(TestPropertyValueCmd):

2375

1805

command = command.SetTimeout

2376

1806

propname = "Timeout"

2377

1807

values_to_set = [datetime.timedelta(),

2382

1812

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2383

1813

2384

1814

2385

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

1815

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

2386

1816

command = command.SetExtendedTimeout

2387

1817

propname = "ExtendedTimeout"

2388

1818

values_to_set = [datetime.timedelta(),

2393

1823

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2394

1824

2395

1825

2396

class TestSetIntervalCmd(TestPropertySetterCmd):

1826

class TestSetIntervalCmd(TestPropertyValueCmd):

2397

1827

command = command.SetInterval

2398

1828

propname = "Interval"

2399

1829

values_to_set = [datetime.timedelta(),

2404

1834

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2405

1835

2406

1836

2407

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

1837

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

2408

1838

command = command.SetApprovalDelay

2409

1839

propname = "ApprovalDelay"

2410

1840

values_to_set = [datetime.timedelta(),

2415

1845

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2416

1846

2417

1847

2418

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

1848

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

2419

1849

command = command.SetApprovalDuration

2420

1850

propname = "ApprovalDuration"

2421

1851

values_to_set = [datetime.timedelta(),

Older »