/mandos/release : revision 237.7.672

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-07-25 22:44:36 UTC
mto: This revision was merged to the branch mainline in revision 384.
Revision ID: teddy@recompile.se-20190725224436-19eyd06r7qmtowo6

Allow line breaks on long lines in plugin-runner manual

* debian/mandos-client.lintian-overrides
  (manpage-has-errors-from-man): Remove.
* plugin-runner.xml (EXAMPLE): Allow line breaks by adding ZERO WIDTH
                               SPACE characters.  Also correct
                               description of example command line.

Show diffs side-by-side

added added

removed removed

mandos-ctl

#!/usr/bin/python

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

# -*- 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 -*-

# Mandos Monitor - Control and monitor the Mandos server

import tempfile

import contextlib

import dbus

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

# Show warnings by default

if not sys.warnoptions:

if sys.version_info.major == 2:

str = unicode

import StringIO

io.StringIO = StringIO.StringIO

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"

version = "1.8.4"

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)

bus = dbus.SystemBus()

mandos_dbus_object = get_mandos_dbus_object(bus)

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

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)

113

110

114

111

# Compile dict of (clientpath: properties) to process

115

112

if not clientnames:

128

125

commands = commands_from_options(options)

129

126

130

127

for command in commands:

131

command.run(clients, bus, mandos_serv)

128

command.run(clients, bus)

132

129

133

130

134

131

def add_command_line_options(parser):

139

136

help="Select all clients")

140

137

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

141

138

help="Print all fields")

142

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

139

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

140

action="append_const", default=[],

141

const=command.DumpJSON(),

143

142

help="Dump client data in JSON format")

144

143

enable_disable = parser.add_mutually_exclusive_group()

145

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

144

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

145

action="append_const", default=[],

146

const=command.Enable(),

146

147

help="Enable client")

147

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

148

action="store_true",

148

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

149

action="append_const", default=[],

150

const=command.Disable(),

149

151

help="disable client")

150

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

152

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

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

151

155

help="Bump timeout for client")

152

156

start_stop_checker = parser.add_mutually_exclusive_group()

153

157

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

154

action="store_true",

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

155

161

help="Start checker for client")

156

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

157

action="store_true",

162

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

163

action="append_const", default=[],

164

const=command.StopChecker(),

158

165

help="Stop checker for client")

159

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

166

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

167

action="append_const", default=[],

168

const=command.IsEnabled(),

160

169

help="Check if client is enabled")

161

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

170

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

171

action="append_const", default=[],

172

const=command.Remove(),

162

173

help="Remove client")

163

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

174

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

175

action="append", default=[],

176

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

164

177

help="Set checker command for client")

165

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

166

help="Set timeout for client")

167

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

168

help="Set extended timeout for client")

169

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

170

help="Set checker interval 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")

171

193

approve_deny_default = parser.add_mutually_exclusive_group()

172

194

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

195

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

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

175

198

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

176

199

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

200

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

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

179

203

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

180

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

181

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

182

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

183

help="Set duration of one client approval")

184

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

185

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

186

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

187

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

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")

188

223

approve_deny = parser.add_mutually_exclusive_group()

189

224

approve_deny.add_argument(

190

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

225

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

226

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

191

227

help="Approve any current client request")

192

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

228

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

229

action="append_const", default=[],

230

const=command.Deny(),

193

231

help="Deny any current client request")

194

232

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

195

233

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

386

424

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

387

425

argparse"""

388

426

389

def has_actions(options):

390

return any((options.enable,

391

options.disable,

392

options.bump_timeout,

393

options.start_checker,

394

options.stop_checker,

395

options.is_enabled,

396

options.remove,

397

options.checker is not None,

398

options.timeout is not None,

399

options.extended_timeout is not None,

400

options.interval is not None,

401

options.approved_by_default is not None,

402

options.approval_delay is not None,

403

options.approval_duration is not None,

404

options.host is not None,

405

options.secret is not None,

406

options.approve,

407

options.deny))

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)

408

450

409

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

451

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

410

452

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

411

if options.verbose and has_actions(options):

453

if options.verbose and has_commands(options):

412

454

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

455

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

456

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

415

457

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

416

if options.all and not has_actions(options):

458

if options.all and not has_commands(options):

417

459

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

418

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

460

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

461

and len(options.client) > 1):

419

462

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

420

if options.remove:

421

options.remove = False

422

if has_actions(options) and not options.deny:

423

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

424

options.remove = True

425

426

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

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

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

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

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

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

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

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

472

self.logger.removeFilter(self.nullfilter)

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

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

575

if key == "Secret":

576

if not isinstance(value, bytes):

577

value = value.encode("utf-8")

578

value = self.dbus_python.ByteArray(value)

579

return self.set_property(self.busname, objectpath,

580

self.client_interface, key,

581

value)

582

583

class SilenceLogger(object):

584

"Simple context manager to silence a particular logger"

585

def __init__(self, loggername):

586

self.logger = logging.getLogger(loggername)

587

588

def __enter__(self):

589

self.logger.addFilter(self.nullfilter)

590

591

class NullFilter(logging.Filter):

592

def filter(self, record):

593

return False

594

595

nullfilter = NullFilter()

596

597

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

598

self.logger.removeFilter(self.nullfilter)

599

600

601

class CachingBus(SystemBus):

602

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

603

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

604

self.object_cache = {}

605

super(dbus_python_adapter.CachingBus,

606

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

607

def get_object(self, busname, objectpath):

608

try:

609

return self.object_cache[(busname, objectpath)]

610

except KeyError:

611

new_object = super(

612

dbus_python_adapter.CachingBus,

613

self).get_object(busname, objectpath)

614

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

615

return new_object

616

617

618

class pydbus_adapter(object):

619

class SystemBus(dbus.MandosBus):

620

def __init__(self, module=pydbus):

621

self.pydbus = module

622

self.bus = self.pydbus.SystemBus()

623

624

@contextlib.contextmanager

625

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

626

try:

627

yield

628

except gi.repository.GLib.Error as e:

629

# This does what "raise from" would do

630

exc = exception_class(*e.args)

631

exc.__cause__ = e

632

raise exc

633

634

def call_method(self, methodname, busname, objectpath,

635

interface, *args):

636

proxy_object = self.get(busname, objectpath)

637

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

638

interface, methodname,

639

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

640

method = getattr(proxy_object[interface], methodname)

641

with self.convert_exception():

642

return method(*args)

643

644

def get(self, busname, objectpath):

645

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

646

busname, objectpath)

647

with self.convert_exception(dbus.ConnectFailed):

648

if sys.version_info.major <= 2:

649

with warnings.catch_warnings():

650

warnings.filterwarnings(

651

"ignore", "", DeprecationWarning,

652

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

653

return self.bus.get(busname, objectpath)

654

else:

655

return self.bus.get(busname, objectpath)

656

657

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

658

value):

659

proxy_object = self.get(busname, objectpath)

660

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

661

objectpath, self.properties_iface, interface,

662

key, value)

663

setattr(proxy_object[interface], key, value)

664

665

class CachingBus(SystemBus):

666

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

667

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

668

self.object_cache = {}

669

super(pydbus_adapter.CachingBus,

670

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

671

def get(self, busname, objectpath):

672

try:

673

return self.object_cache[(busname, objectpath)]

674

except KeyError:

675

new_object = (super(pydbus_adapter.CachingBus, self)

676

.get(busname, objectpath))

677

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

678

return new_object

473

679

474

680

475

681

def commands_from_options(options):

476

682

477

commands = []

478

479

if options.is_enabled:

480

commands.append(command.IsEnabled())

481

482

if options.approve:

483

commands.append(command.Approve())

484

485

if options.deny:

486

commands.append(command.Deny())

487

488

if options.remove:

489

commands.append(command.Remove())

490

491

if options.dump_json:

492

commands.append(command.DumpJSON())

493

494

if options.enable:

495

commands.append(command.Enable())

496

497

if options.disable:

498

commands.append(command.Disable())

499

500

if options.bump_timeout:

501

commands.append(command.BumpTimeout())

502

503

if options.start_checker:

504

commands.append(command.StartChecker())

505

506

if options.stop_checker:

507

commands.append(command.StopChecker())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(command.ApproveByDefault())

683

commands = list(options.commands)

684

685

def find_cmd(cmd, commands):

686

i = 0

687

for i, c in enumerate(commands):

688

if isinstance(c, cmd):

689

return i

690

return i+1

691

692

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

693

# to occur ahead of command.Remove.

694

index_of_remove = find_cmd(command.Remove, commands)

695

before_remove = commands[:index_of_remove]

696

after_remove = commands[index_of_remove:]

697

cleaned_after = []

698

for cmd in after_remove:

699

if isinstance(cmd, command.Deny):

700

before_remove.append(cmd)

512

701

else:

513

commands.append(command.DenyByDefault())

514

515

if options.checker is not None:

516

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

517

518

if options.host is not None:

519

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

520

521

if options.secret is not None:

522

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

523

524

if options.timeout is not None:

525

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

526

527

if options.extended_timeout:

528

commands.append(

529

command.SetExtendedTimeout(options.extended_timeout))

530

531

if options.interval is not None:

532

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

533

534

if options.approval_delay is not None:

535

commands.append(

536

command.SetApprovalDelay(options.approval_delay))

537

538

if options.approval_duration is not None:

539

commands.append(

540

command.SetApprovalDuration(options.approval_duration))

702

cleaned_after.append(cmd)

703

if cleaned_after != after_remove:

704

commands = before_remove + cleaned_after

541

705

542

706

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

543

707

# optionally verbosely

552

716

553

717

class Base(object):

554

718

"""Abstract base class for commands"""

555

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

719

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

556

720

"""Normal commands should implement run_on_one_client(),

557

721

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

558

722

override this run() method instead.

559

723

"""

560

self.mandos = mandos

561

for clientpath, properties in clients.items():

562

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

563

dbus_busname, str(clientpath))

564

client = bus.get_object(dbus_busname, clientpath)

724

self.bus = bus

725

for client, properties in clients.items():

565

726

self.run_on_one_client(client, properties)

566

727

567

728

568

729

class IsEnabled(Base):

569

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

570

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

571

if self.is_enabled(client, properties):

730

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

731

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

732

if properties["Enabled"]:

572

733

sys.exit(0)

573

734

sys.exit(1)

574

def is_enabled(self, client, properties):

575

return properties["Enabled"]

576

735

577

736

578

737

class Approve(Base):

579

738

def run_on_one_client(self, client, properties):

580

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

581

client.__dbus_object_path__,

582

client_dbus_interface)

583

client.Approve(dbus.Boolean(True),

584

dbus_interface=client_dbus_interface)

739

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

585

740

586

741

587

742

class Deny(Base):

588

743

def run_on_one_client(self, client, properties):

589

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

590

client.__dbus_object_path__,

591

client_dbus_interface)

592

client.Approve(dbus.Boolean(False),

593

dbus_interface=client_dbus_interface)

744

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

594

745

595

746

596

747

class Remove(Base):

597

def run_on_one_client(self, client, properties):

598

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

599

dbus_busname, server_dbus_path,

600

server_dbus_interface,

601

str(client.__dbus_object_path__))

602

self.mandos.RemoveClient(client.__dbus_object_path__)

748

def run(self, clients, bus):

749

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

750

bus.call_server_method("RemoveClient", clientpath)

603

751

604

752

605

753

class Output(Base):

613

761

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

614

762

"LastCheckerStatus")

615

763

616

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

617

print(self.output(clients.values()))

618

619

def output(self, clients):

620

raise NotImplementedError()

621

622

764

623

765

class DumpJSON(Output):

624

def output(self, clients):

625

data = {client["Name"]:

626

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

766

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

767

data = {properties["Name"]:

768

{key: properties[key]

627

769

for key in self.all_keywords}

628

for client in clients}

629

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

630

631

@staticmethod

632

def dbus_boolean_to_bool(value):

633

if isinstance(value, dbus.Boolean):

634

value = bool(value)

635

return value

770

for properties in clients.values()}

771

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

636

772

637

773

638

774

class PrintTable(Output):

639

775

def __init__(self, verbose=False):

640

776

self.verbose = verbose

641

777

642

def output(self, clients):

778

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

643

779

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

644

780

"LastCheckedOK")

645

781

keywords = default_keywords

646

782

if self.verbose:

647

783

keywords = self.all_keywords

648

return str(self.TableOfClients(clients, keywords))

784

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

649

785

650

786

class TableOfClients(object):

651

787

tableheaders = {

706

842

707

843

@classmethod

708

844

def valuetostring(cls, value, keyword):

709

if isinstance(value, dbus.Boolean):

845

if isinstance(value, bool):

710

846

return "Yes" if value else "No"

711

847

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

712

848

"ApprovalDuration", "ExtendedTimeout"):

732

868

seconds=td.seconds % 60))

733

869

734

870

735

class Property(Base):

871

class PropertySetter(Base):

736

872

"Abstract class for Actions for setting one client property"

737

873

738

def run_on_one_client(self, client, properties):

874

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

739

875

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

740

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

741

client.__dbus_object_path__,

742

dbus.PROPERTIES_IFACE, client_dbus_interface,

743

self.propname, self.value_to_set

744

if not isinstance(self.value_to_set,

745

dbus.Boolean)

746

else bool(self.value_to_set))

747

client.Set(client_dbus_interface, self.propname,

748

self.value_to_set,

749

dbus_interface=dbus.PROPERTIES_IFACE)

876

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

877

self.value_to_set)

750

878

751

879

@property

752

880

def propname(self):

753

881

raise NotImplementedError()

754

882

755

883

756

class Enable(Property):

757

propname = "Enabled"

758

value_to_set = dbus.Boolean(True)

759

760

761

class Disable(Property):

762

propname = "Enabled"

763

value_to_set = dbus.Boolean(False)

764

765

766

class BumpTimeout(Property):

884

class Enable(PropertySetter):

885

propname = "Enabled"

886

value_to_set = True

887

888

889

class Disable(PropertySetter):

890

propname = "Enabled"

891

value_to_set = False

892

893

894

class BumpTimeout(PropertySetter):

767

895

propname = "LastCheckedOK"

768

896

value_to_set = ""

769

897

770

898

771

class StartChecker(Property):

772

propname = "CheckerRunning"

773

value_to_set = dbus.Boolean(True)

774

775

776

class StopChecker(Property):

777

propname = "CheckerRunning"

778

value_to_set = dbus.Boolean(False)

779

780

781

class ApproveByDefault(Property):

782

propname = "ApprovedByDefault"

783

value_to_set = dbus.Boolean(True)

784

785

786

class DenyByDefault(Property):

787

propname = "ApprovedByDefault"

788

value_to_set = dbus.Boolean(False)

789

790

791

class PropertyValue(Property):

792

"Abstract class for Property recieving a value as argument"

899

class StartChecker(PropertySetter):

900

propname = "CheckerRunning"

901

value_to_set = True

902

903

904

class StopChecker(PropertySetter):

905

propname = "CheckerRunning"

906

value_to_set = False

907

908

909

class ApproveByDefault(PropertySetter):

910

propname = "ApprovedByDefault"

911

value_to_set = True

912

913

914

class DenyByDefault(PropertySetter):

915

propname = "ApprovedByDefault"

916

value_to_set = False

917

918

919

class PropertySetterValue(PropertySetter):

920

"""Abstract class for PropertySetter recieving a value as

921

constructor argument instead of a class attribute."""

793

922

def __init__(self, value):

794

923

self.value_to_set = value

795

924

925

@classmethod

926

def argparse(cls, argtype):

927

def cmdtype(arg):

928

return cls(argtype(arg))

929

return cmdtype

796

930

797

class SetChecker(PropertyValue):

931

class SetChecker(PropertySetterValue):

798

932

propname = "Checker"

799

933

800

934

801

class SetHost(PropertyValue):

935

class SetHost(PropertySetterValue):

802

936

propname = "Host"

803

937

804

938

805

class SetSecret(PropertyValue):

939

class SetSecret(PropertySetterValue):

806

940

propname = "Secret"

807

941

808

942

@property

816

950

value.close()

817

951

818

952

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

953

class PropertySetterValueMilliseconds(PropertySetterValue):

954

"""Abstract class for PropertySetterValue taking a value

955

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

956

milliseconds."""

822

957

823

958

@property

824

959

def value_to_set(self):

830

965

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

831

966

832

967

833

class SetTimeout(MillisecondsPropertyValueArgument):

968

class SetTimeout(PropertySetterValueMilliseconds):

834

969

propname = "Timeout"

835

970

836

971

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

972

class SetExtendedTimeout(PropertySetterValueMilliseconds):

838

973

propname = "ExtendedTimeout"

839

974

840

975

841

class SetInterval(MillisecondsPropertyValueArgument):

976

class SetInterval(PropertySetterValueMilliseconds):

842

977

propname = "Interval"

843

978

844

979

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

980

class SetApprovalDelay(PropertySetterValueMilliseconds):

846

981

propname = "ApprovalDelay"

847

982

848

983

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

984

class SetApprovalDuration(PropertySetterValueMilliseconds):

850

985

propname = "ApprovalDuration"

851

986

852

987

885

1020

"output"))

886

1021

887

1022

1023

class Unique(object):

1024

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

1025

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

1026

1027

888

1028

class Test_string_to_delta(TestCaseWithAssertLogs):

889

def test_handles_basic_rfc3339(self):

890

self.assertEqual(string_to_delta("PT0S"),

891

datetime.timedelta())

892

self.assertEqual(string_to_delta("P0D"),

893

datetime.timedelta())

894

self.assertEqual(string_to_delta("PT1S"),

895

datetime.timedelta(0, 1))

896

self.assertEqual(string_to_delta("PT2H"),

897

datetime.timedelta(0, 7200))

1029

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

1030

# rfc3339_duration_to_delta() already has more comprehensive

1031

# tests, which are run by doctest.

1032

1033

def test_rfc3339_zero_seconds(self):

1034

self.assertEqual(datetime.timedelta(),

1035

string_to_delta("PT0S"))

1036

1037

def test_rfc3339_zero_days(self):

1038

self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))

1039

1040

def test_rfc3339_one_second(self):

1041

self.assertEqual(datetime.timedelta(0, 1),

1042

string_to_delta("PT1S"))

1043

1044

def test_rfc3339_two_hours(self):

1045

self.assertEqual(datetime.timedelta(0, 7200),

1046

string_to_delta("PT2H"))

898

1047

899

1048

def test_falls_back_to_pre_1_6_1_with_warning(self):

900

1049

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

901

1050

value = string_to_delta("2h")

902

self.assertEqual(value, datetime.timedelta(0, 7200))

1051

self.assertEqual(datetime.timedelta(0, 7200), value)

903

1052

904

1053

905

1054

class Test_check_option_syntax(unittest.TestCase):

909

1058

910

1059

def test_actions_requires_client_or_all(self):

911

1060

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

912

options = self.parser.parse_args()

913

setattr(options, action, value)

1061

args = self.actionargs(action, value)

914

1062

with self.assertParseError():

915

self.check_option_syntax(options)

1063

self.parse_args(args)

916

1064

917

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

1065

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

918

1066

# check_option_syntax()

919

1067

actions = {

920

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

921

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

922

# never used

923

"enable": True,

924

"disable": True,

925

"bump_timeout": True,

926

"start_checker": True,

927

"stop_checker": True,

928

"is_enabled": True,

929

"remove": True,

930

"checker": "x",

931

"timeout": datetime.timedelta(),

932

"extended_timeout": datetime.timedelta(),

933

"interval": datetime.timedelta(),

934

"approved_by_default": True,

935

"approval_delay": datetime.timedelta(),

936

"approval_duration": datetime.timedelta(),

937

"host": "x",

938

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

939

"approve": True,

940

"deny": True,

1068

"--enable": None,

1069

"--disable": None,

1070

"--bump-timeout": None,

1071

"--start-checker": None,

1072

"--stop-checker": None,

1073

"--is-enabled": None,

1074

"--remove": None,

1075

"--checker": "x",

1076

"--timeout": "PT0S",

1077

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

1078

"--interval": "PT0S",

1079

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

1080

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

1081

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

1082

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

1083

"--host": "hostname",

1084

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

1085

"--approve": None,

1086

"--deny": None,

941

1087

}

942

1088

1089

@staticmethod

1090

def actionargs(action, value, *args):

1091

if value is not None:

1092

return [action, value] + list(args)

1093

else:

1094

return [action] + list(args)

1095

943

1096

@contextlib.contextmanager

944

1097

def assertParseError(self):

945

1098

with self.assertRaises(SystemExit) as e:

948

1101

# Exit code from argparse is guaranteed to be "2". Reference:

949

1102

# https://docs.python.org/3/library

950

1103

# /argparse.html#exiting-methods

951

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

1104

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

1105

1106

def parse_args(self, args):

1107

options = self.parser.parse_args(args)

1108

check_option_syntax(self.parser, options)

952

1109

953

1110

@staticmethod

954

1111

@contextlib.contextmanager

955

1112

def redirect_stderr_to_devnull():

956

null = os.open(os.path.devnull, os.O_RDWR)

957

stderrcopy = os.dup(sys.stderr.fileno())

958

os.dup2(null, sys.stderr.fileno())

959

os.close(null)

960

try:

961

yield

962

finally:

963

# restore stderr

964

os.dup2(stderrcopy, sys.stderr.fileno())

965

os.close(stderrcopy)

1113

old_stderr = sys.stderr

1114

with contextlib.closing(open(os.devnull, "w")) as null:

1115

sys.stderr = null

1116

try:

1117

yield

1118

finally:

1119

sys.stderr = old_stderr

966

1120

967

1121

def check_option_syntax(self, options):

968

1122

check_option_syntax(self.parser, options)

969

1123

970

1124

def test_actions_all_conflicts_with_verbose(self):

971

1125

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

972

options = self.parser.parse_args()

973

setattr(options, action, value)

974

options.all = True

975

options.verbose = True

1126

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

1127

"--verbose")

976

1128

with self.assertParseError():

977

self.check_option_syntax(options)

1129

self.parse_args(args)

978

1130

979

1131

def test_actions_with_client_conflicts_with_verbose(self):

980

1132

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

981

options = self.parser.parse_args()

982

setattr(options, action, value)

983

options.verbose = True

984

options.client = ["foo"]

1133

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

1134

"client")

985

1135

with self.assertParseError():

986

self.check_option_syntax(options)

1136

self.parse_args(args)

987

1137

988

1138

def test_dump_json_conflicts_with_verbose(self):

989

options = self.parser.parse_args()

990

options.dump_json = True

991

options.verbose = True

1139

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

992

1140

with self.assertParseError():

993

self.check_option_syntax(options)

1141

self.parse_args(args)

994

1142

995

1143

def test_dump_json_conflicts_with_action(self):

996

1144

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

997

options = self.parser.parse_args()

998

setattr(options, action, value)

999

options.dump_json = True

1145

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

1000

1146

with self.assertParseError():

1001

self.check_option_syntax(options)

1147

self.parse_args(args)

1002

1148

1003

1149

def test_all_can_not_be_alone(self):

1004

options = self.parser.parse_args()

1005

options.all = True

1150

args = ["--all"]

1006

1151

with self.assertParseError():

1007

self.check_option_syntax(options)

1152

self.parse_args(args)

1008

1153

1009

1154

def test_all_is_ok_with_any_action(self):

1010

1155

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

1011

options = self.parser.parse_args()

1012

setattr(options, action, value)

1013

options.all = True

1014

self.check_option_syntax(options)

1156

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

1157

self.parse_args(args)

1015

1158

1016

1159

def test_any_action_is_ok_with_one_client(self):

1017

1160

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

1018

options = self.parser.parse_args()

1019

setattr(options, action, value)

1020

options.client = ["foo"]

1021

self.check_option_syntax(options)

1161

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

1162

self.parse_args(args)

1022

1163

1023

1164

def test_one_client_with_all_actions_except_is_enabled(self):

1024

options = self.parser.parse_args()

1025

1165

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

1026

if action == "is_enabled":

1166

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

1027

1167

continue

1028

setattr(options, action, value)

1029

options.client = ["foo"]

1030

self.check_option_syntax(options)

1168

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

1169

self.parse_args(args)

1031

1170

1032

1171

def test_two_clients_with_all_actions_except_is_enabled(self):

1033

options = self.parser.parse_args()

1034

1172

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

1035

if action == "is_enabled":

1173

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

1036

1174

continue

1037

setattr(options, action, value)

1038

options.client = ["foo", "barbar"]

1039

self.check_option_syntax(options)

1175

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

1176

"client2")

1177

self.parse_args(args)

1040

1178

1041

1179

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1042

1180

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

1043

if action == "is_enabled":

1181

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

1044

1182

continue

1045

options = self.parser.parse_args()

1046

setattr(options, action, value)

1047

options.client = ["foo", "barbar"]

1048

self.check_option_syntax(options)

1183

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

1184

"client2")

1185

self.parse_args(args)

1049

1186

1050

1187

def test_is_enabled_fails_without_client(self):

1051

options = self.parser.parse_args()

1052

options.is_enabled = True

1188

args = ["--is-enabled"]

1053

1189

with self.assertParseError():

1054

self.check_option_syntax(options)

1190

self.parse_args(args)

1055

1191

1056

1192

def test_is_enabled_fails_with_two_clients(self):

1057

options = self.parser.parse_args()

1058

options.is_enabled = True

1059

options.client = ["foo", "barbar"]

1193

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

1060

1194

with self.assertParseError():

1061

self.check_option_syntax(options)

1195

self.parse_args(args)

1062

1196

1063

1197

def test_remove_can_only_be_combined_with_action_deny(self):

1064

1198

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

1065

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

1199

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

1066

1200

continue

1067

options = self.parser.parse_args()

1068

setattr(options, action, value)

1069

options.all = True

1070

options.remove = True

1201

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

1202

"--remove")

1071

1203

with self.assertParseError():

1072

self.check_option_syntax(options)

1073

1074

1075

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1076

def test_calls_and_returns_get_object_on_bus(self):

1077

class MockBus(object):

1078

called = False

1079

def get_object(mockbus_self, busname, dbus_path):

1080

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

1081

# this MockBus instance is in "mockbus_self".

1082

self.assertEqual(busname, dbus_busname)

1083

self.assertEqual(dbus_path, server_dbus_path)

1084

mockbus_self.called = True

1085

return mockbus_self

1086

1087

mockbus = get_mandos_dbus_object(bus=MockBus())

1088

self.assertIsInstance(mockbus, MockBus)

1089

self.assertTrue(mockbus.called)

1090

1091

def test_logs_and_exits_on_dbus_error(self):

1092

class MockBusFailing(object):

1093

def get_object(self, busname, dbus_path):

1094

raise dbus.exceptions.DBusException("Test")

1095

1096

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

1097

with self.assertRaises(SystemExit) as e:

1098

bus = get_mandos_dbus_object(bus=MockBusFailing())

1099

1100

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

1101

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

1102

else:

1103

self.assertIsNotNone(e.exception.code)

1104

1105

1106

class Test_get_managed_objects(TestCaseWithAssertLogs):

1107

def test_calls_and_returns_GetManagedObjects(self):

1108

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

1109

class MockObjectManager(object):

1110

def GetManagedObjects(self):

1111

return managed_objects

1112

retval = get_managed_objects(MockObjectManager())

1113

self.assertDictEqual(managed_objects, retval)

1114

1115

def test_logs_and_exits_on_dbus_error(self):

1204

self.parse_args(args)

1205

1206

1207

class Test_dbus_exceptions(unittest.TestCase):

1208

1209

def test_dbus_ConnectFailed_is_Error(self):

1210

with self.assertRaises(dbus.Error):

1211

raise dbus.ConnectFailed()

1212

1213

1214

class Test_dbus_MandosBus(unittest.TestCase):

1215

1216

class MockMandosBus(dbus.MandosBus):

1217

def __init__(self):

1218

self._name = "se.recompile.Mandos"

1219

self._server_path = "/"

1220

self._server_interface = "se.recompile.Mandos"

1221

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

1222

self.calls = []

1223

self.call_method_return = Unique()

1224

1225

def call_method(self, methodname, busname, objectpath,

1226

interface, *args):

1227

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

1228

interface, args))

1229

return self.call_method_return

1230

1231

def setUp(self):

1232

self.bus = self.MockMandosBus()

1233

1234

def test_set_client_property(self):

1235

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

1236

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

1237

"org.freedesktop.DBus.Properties",

1238

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

1239

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

1240

1241

def test_call_client_method(self):

1242

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

1243

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

1244

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

1245

self.bus._client_interface, ())

1246

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

1247

1248

def test_call_client_method_with_args(self):

1249

args = (Unique(), Unique())

1250

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

1251

*args)

1252

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

1253

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

1254

self.bus._client_interface,

1255

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

1256

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

1257

1258

def test_get_clients_and_properties(self):

1259

managed_objects = {

1260

"objectpath": {

1261

self.bus._client_interface: {

1262

"key": "value",

1263

"bool": True,

1264

1265

"irrelevant_interface": {

1266

"key": "othervalue",

1267

"bool": False,

1268

1269

1270

"other_objectpath": {

1271

"other_irrelevant_interface": {

1272

"key": "value 3",

1273

"bool": None,

1274

1275

1276

}

1277

expected_clients_and_properties = {

1278

"objectpath": {

1279

"key": "value",

1280

"bool": True,

1281

}

1282

}

1283

self.bus.call_method_return = managed_objects

1284

ret = self.bus.get_clients_and_properties()

1285

self.assertDictEqual(expected_clients_and_properties, ret)

1286

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

1287

self.bus._server_path,

1288

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

1289

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

1290

1291

def test_call_server_method(self):

1292

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

1293

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

1294

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

1295

self.bus._server_path,

1296

self.bus._server_interface, ())

1297

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

1298

1299

def test_call_server_method_with_args(self):

1300

args = (Unique(), Unique())

1301

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

1302

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

1303

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

1304

self.bus._server_path,

1305

self.bus._server_interface,

1306

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

1307

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

1308

1309

1310

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1311

1312

def MockDBusPython_func(self, func):

1313

class mock_dbus_python(object):

1314

"""mock dbus-python module"""

1315

class exceptions(object):

1316

"""Pseudo-namespace"""

1317

class DBusException(Exception):

1318

pass

1319

class SystemBus(object):

1320

@staticmethod

1321

def get_object(busname, objectpath):

1322

DBusObject = collections.namedtuple(

1323

"DBusObject", ("methodname", "Set"))

1324

def method(*args, **kwargs):

1325

self.assertEqual({"dbus_interface":

1326

"interface"},

1327

kwargs)

1328

return func(*args)

1329

def set_property(interface, key, value,

1330

dbus_interface=None):

1331

self.assertEqual(

1332

"org.freedesktop.DBus.Properties",

1333

dbus_interface)

1334

self.assertEqual("Secret", key)

1335

return func(interface, key, value,

1336

dbus_interface=dbus_interface)

1337

return DBusObject(methodname=method,

1338

Set=set_property)

1339

class Boolean(object):

1340

def __init__(self, value):

1341

self.value = bool(value)

1342

def __bool__(self):

1343

return self.value

1344

if sys.version_info.major == 2:

1345

__nonzero__ = __bool__

1346

class ObjectPath(str):

1347

pass

1348

class Dictionary(dict):

1349

pass

1350

class ByteArray(bytes):

1351

pass

1352

return mock_dbus_python

1353

1354

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

1355

interface, *args):

1356

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

1357

return bus.call_method(methodname, busname, objectpath,

1358

interface, *args)

1359

1360

def test_call_method_returns(self):

1361

expected_method_return = Unique()

1362

method_args = (Unique(), Unique())

1363

def func(*args):

1364

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

1365

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

1366

self.assertIs(marg, arg)

1367

return expected_method_return

1368

mock_dbus_python = self.MockDBusPython_func(func)

1369

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1370

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

1371

"objectpath", "interface",

1372

*method_args)

1373

self.assertIs(ret, expected_method_return)

1374

1375

def test_call_method_filters_bool_true(self):

1376

def func():

1377

return method_return

1378

mock_dbus_python = self.MockDBusPython_func(func)

1379

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1380

method_return = mock_dbus_python.Boolean(True)

1381

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

1382

"objectpath", "interface")

1383

self.assertTrue(ret)

1384

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1385

1386

def test_call_method_filters_bool_false(self):

1387

def func():

1388

return method_return

1389

mock_dbus_python = self.MockDBusPython_func(func)

1390

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1391

method_return = mock_dbus_python.Boolean(False)

1392

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

1393

"objectpath", "interface")

1394

self.assertFalse(ret)

1395

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1396

1397

def test_call_method_filters_objectpath(self):

1398

def func():

1399

return method_return

1400

mock_dbus_python = self.MockDBusPython_func(func)

1401

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1402

method_return = mock_dbus_python.ObjectPath("objectpath")

1403

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

1404

"objectpath", "interface")

1405

self.assertEqual("objectpath", ret)

1406

self.assertIsNot("objectpath", ret)

1407

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1408

1409

def test_call_method_filters_booleans_in_dict(self):

1410

def func():

1411

return method_return

1412

mock_dbus_python = self.MockDBusPython_func(func)

1413

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1414

method_return = mock_dbus_python.Dictionary(

1415

{mock_dbus_python.Boolean(True):

1416

mock_dbus_python.Boolean(False),

1417

mock_dbus_python.Boolean(False):

1418

mock_dbus_python.Boolean(True)})

1419

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

1420

"objectpath", "interface")

1421

expected_method_return = {True: False,

1422

False: True}

1423

self.assertEqual(expected_method_return, ret)

1424

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1425

1426

def test_call_method_filters_objectpaths_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

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1433

mock_dbus_python.ObjectPath("objectpath_value_1"),

1434

mock_dbus_python.ObjectPath("objectpath_key_2"):

1435

mock_dbus_python.ObjectPath("objectpath_value_2")})

1436

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

1437

"objectpath", "interface")

1438

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

1439

for key, value in

1440

method_return.items()}

1441

self.assertEqual(expected_method_return, ret)

1442

self.assertIsInstance(ret, dict)

1443

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1444

1445

def test_call_method_filters_dict_in_dict(self):

1446

def func():

1447

return method_return

1448

mock_dbus_python = self.MockDBusPython_func(func)

1449

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1450

method_return = mock_dbus_python.Dictionary(

1451

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

1452

"key12": "value12"}),

1453

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

1454

"key22": "value22"})})

1455

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

1456

"objectpath", "interface")

1457

expected_method_return = {

1458

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

1459

"key12": "value12"},

1460

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

1461

"key22": "value22"},

1462

}

1463

self.assertEqual(expected_method_return, ret)

1464

self.assertIsInstance(ret, dict)

1465

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1466

for key, value in ret.items():

1467

self.assertIsInstance(value, dict)

1468

self.assertEqual(expected_method_return[key], value)

1469

self.assertNotIsInstance(value,

1470

mock_dbus_python.Dictionary)

1471

1472

def test_call_method_filters_dict_three_deep(self):

1473

def func():

1474

return method_return

1475

mock_dbus_python = self.MockDBusPython_func(func)

1476

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1477

method_return = mock_dbus_python.Dictionary(

1478

{"key1":

1479

mock_dbus_python.Dictionary(

1480

{"key2":

1481

mock_dbus_python.Dictionary(

1482

{"key3":

1483

mock_dbus_python.Boolean(True),

1484

}),

1485

}),

1486

})

1487

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

1488

"objectpath", "interface")

1489

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

1490

self.assertEqual(expected_method_return, ret)

1491

self.assertIsInstance(ret, dict)

1492

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1493

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

1494

self.assertNotIsInstance(ret["key1"],

1495

mock_dbus_python.Dictionary)

1496

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

1497

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

1498

mock_dbus_python.Dictionary)

1499

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

1500

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

1501

mock_dbus_python.Boolean)

1502

1503

def test_call_method_handles_exception(self):

1116

1504

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

1117

1505

1118

class MockObjectManagerFailing(object):

1119

def GetManagedObjects(self):

1120

dbus_logger.error("Test")

1121

raise dbus.exceptions.DBusException("Test")

1506

def func():

1507

dbus_logger.error("Test")

1508

raise mock_dbus_python.exceptions.DBusException()

1509

1510

mock_dbus_python = self.MockDBusPython_func(func)

1511

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1122

1512

1123

1513

class CountingHandler(logging.Handler):

1124

1514

count = 0

1130

1520

dbus_logger.addHandler(counting_handler)

1131

1521

1132

1522

try:

1133

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

1134

with self.assertRaises(SystemExit) as e:

1135

get_managed_objects(MockObjectManagerFailing())

1523

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

1524

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

1525

"objectpath", "interface")

1136

1526

finally:

1137

1527

dbus_logger.removeFilter(counting_handler)

1138

1528

1529

self.assertNotIsInstance(e, dbus.ConnectFailed)

1530

1139

1531

# Make sure the dbus logger was suppressed

1140

self.assertEqual(counting_handler.count, 0)

1141

1142

# Test that the dbus_logger still works

1143

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

1144

dbus_logger.error("Test")

1145

1146

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

1147

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

1148

else:

1149

self.assertIsNotNone(e.exception.code)

1532

self.assertEqual(0, counting_handler.count)

1533

1534

def test_Set_Secret_sends_bytearray(self):

1535

ret = [None]

1536

def func(*args, **kwargs):

1537

ret[0] = (args, kwargs)

1538

mock_dbus_python = self.MockDBusPython_func(func)

1539

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1540

bus.set_client_property("objectpath", "Secret", "value")

1541

expected_call = (("se.recompile.Mandos.Client", "Secret",

1542

mock_dbus_python.ByteArray(b"value")),

1543

{"dbus_interface":

1544

"org.freedesktop.DBus.Properties"})

1545

self.assertEqual(expected_call, ret[0])

1546

if sys.version_info.major == 2:

1547

self.assertIsInstance(ret[0][0][-1],

1548

mock_dbus_python.ByteArray)

1549

1550

def test_get_object_converts_to_correct_exception(self):

1551

bus = dbus_python_adapter.SystemBus(

1552

self.fake_dbus_python_raises_exception_on_connect)

1553

with self.assertRaises(dbus.ConnectFailed):

1554

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

1555

"objectpath", "interface")

1556

1557

class fake_dbus_python_raises_exception_on_connect(object):

1558

"""fake dbus-python module"""

1559

class exceptions(object):

1560

"""Pseudo-namespace"""

1561

class DBusException(Exception):

1562

pass

1563

1564

@classmethod

1565

def SystemBus(cls):

1566

def get_object(busname, objectpath):

1567

raise cls.exceptions.DBusException()

1568

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

1569

return Bus(get_object=get_object)

1570

1571

1572

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1573

class mock_dbus_python(object):

1574

"""mock dbus-python modules"""

1575

class SystemBus(object):

1576

@staticmethod

1577

def get_object(busname, objectpath):

1578

return Unique()

1579

1580

def setUp(self):

1581

self.bus = dbus_python_adapter.CachingBus(

1582

self.mock_dbus_python)

1583

1584

def test_returns_distinct_objectpaths(self):

1585

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

1586

self.assertIsInstance(obj1, Unique)

1587

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

1588

self.assertIsInstance(obj2, Unique)

1589

self.assertIsNot(obj1, obj2)

1590

1591

def test_returns_distinct_busnames(self):

1592

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

1593

self.assertIsInstance(obj1, Unique)

1594

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

1595

self.assertIsInstance(obj2, Unique)

1596

self.assertIsNot(obj1, obj2)

1597

1598

def test_returns_distinct_both(self):

1599

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

1600

self.assertIsInstance(obj1, Unique)

1601

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

1602

self.assertIsInstance(obj2, Unique)

1603

self.assertIsNot(obj1, obj2)

1604

1605

def test_returns_same(self):

1606

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

1607

self.assertIsInstance(obj1, Unique)

1608

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

1609

self.assertIsInstance(obj2, Unique)

1610

self.assertIs(obj1, obj2)

1611

1612

def test_returns_same_old(self):

1613

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

1614

self.assertIsInstance(obj1, Unique)

1615

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

1616

self.assertIsInstance(obj2, Unique)

1617

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

1618

self.assertIsInstance(obj1b, Unique)

1619

self.assertIsNot(obj1, obj2)

1620

self.assertIsNot(obj2, obj1b)

1621

self.assertIs(obj1, obj1b)

1622

1623

1624

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1625

1626

def Stub_pydbus_func(self, func):

1627

class stub_pydbus(object):

1628

"""stub pydbus module"""

1629

class SystemBus(object):

1630

@staticmethod

1631

def get(busname, objectpath):

1632

DBusObject = collections.namedtuple(

1633

"DBusObject", ("methodname",))

1634

return {"interface":

1635

DBusObject(methodname=func)}

1636

return stub_pydbus

1637

1638

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

1639

interface, *args):

1640

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

1641

return bus.call_method(methodname, busname, objectpath,

1642

interface, *args)

1643

1644

def test_call_method_returns(self):

1645

expected_method_return = Unique()

1646

method_args = (Unique(), Unique())

1647

def func(*args):

1648

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

1649

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

1650

self.assertIs(marg, arg)

1651

return expected_method_return

1652

stub_pydbus = self.Stub_pydbus_func(func)

1653

bus = pydbus_adapter.SystemBus(stub_pydbus)

1654

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

1655

"objectpath", "interface",

1656

*method_args)

1657

self.assertIs(ret, expected_method_return)

1658

1659

def test_call_method_handles_exception(self):

1660

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

1661

1662

def func():

1663

raise gi.repository.GLib.Error()

1664

1665

stub_pydbus = self.Stub_pydbus_func(func)

1666

bus = pydbus_adapter.SystemBus(stub_pydbus)

1667

1668

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

1669

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

1670

"objectpath", "interface")

1671

1672

self.assertNotIsInstance(e, dbus.ConnectFailed)

1673

1674

def test_get_converts_to_correct_exception(self):

1675

bus = pydbus_adapter.SystemBus(

1676

self.fake_pydbus_raises_exception_on_connect)

1677

with self.assertRaises(dbus.ConnectFailed):

1678

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

1679

"objectpath", "interface")

1680

1681

class fake_pydbus_raises_exception_on_connect(object):

1682

"""fake dbus-python module"""

1683

@classmethod

1684

def SystemBus(cls):

1685

def get(busname, objectpath):

1686

raise gi.repository.GLib.Error()

1687

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

1688

return Bus(get=get)

1689

1690

def test_set_property_uses_setattr(self):

1691

class Object(object):

1692

pass

1693

obj = Object()

1694

class pydbus_spy(object):

1695

class SystemBus(object):

1696

@staticmethod

1697

def get(busname, objectpath):

1698

return {"interface": obj}

1699

bus = pydbus_adapter.SystemBus(pydbus_spy)

1700

value = Unique()

1701

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

1702

value)

1703

self.assertIs(value, obj.key)

1704

1705

def test_get_suppresses_xml_deprecation_warning(self):

1706

if sys.version_info.major >= 3:

1707

return

1708

class stub_pydbus_get(object):

1709

class SystemBus(object):

1710

@staticmethod

1711

def get(busname, objectpath):

1712

warnings.warn_explicit(

1713

"deprecated", DeprecationWarning,

1714

"xml.etree.ElementTree", 0)

1715

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1716

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

1717

warnings.simplefilter("always")

1718

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

1719

self.assertEqual(0, len(w))

1720

1721

1722

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1723

class stub_pydbus(object):

1724

"""stub pydbus module"""

1725

class SystemBus(object):

1726

@staticmethod

1727

def get(busname, objectpath):

1728

return Unique()

1729

1730

def setUp(self):

1731

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

1732

1733

def test_returns_distinct_objectpaths(self):

1734

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

1735

self.assertIsInstance(obj1, Unique)

1736

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

1737

self.assertIsInstance(obj2, Unique)

1738

self.assertIsNot(obj1, obj2)

1739

1740

def test_returns_distinct_busnames(self):

1741

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

1742

self.assertIsInstance(obj1, Unique)

1743

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

1744

self.assertIsInstance(obj2, Unique)

1745

self.assertIsNot(obj1, obj2)

1746

1747

def test_returns_distinct_both(self):

1748

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

1749

self.assertIsInstance(obj1, Unique)

1750

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

1751

self.assertIsInstance(obj2, Unique)

1752

self.assertIsNot(obj1, obj2)

1753

1754

def test_returns_same(self):

1755

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

1756

self.assertIsInstance(obj1, Unique)

1757

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

1758

self.assertIsInstance(obj2, Unique)

1759

self.assertIs(obj1, obj2)

1760

1761

def test_returns_same_old(self):

1762

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

1763

self.assertIsInstance(obj1, Unique)

1764

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

1765

self.assertIsInstance(obj2, Unique)

1766

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

1767

self.assertIsInstance(obj1b, Unique)

1768

self.assertIsNot(obj1, obj2)

1769

self.assertIsNot(obj2, obj1b)

1770

self.assertIs(obj1, obj1b)

1150

1771

1151

1772

1152

1773

class Test_commands_from_options(unittest.TestCase):

1774

1153

1775

def setUp(self):

1154

1776

self.parser = argparse.ArgumentParser()

1155

1777

add_command_line_options(self.parser)

1156

1778

1157

1779

def test_is_enabled(self):

1158

self.assert_command_from_args(["--is-enabled", "foo"],

1780

self.assert_command_from_args(["--is-enabled", "client"],

1159

1781

command.IsEnabled)

1160

1782

1161

def assert_command_from_args(self, args, command_cls,

1162

**cmd_attrs):

1783

def assert_command_from_args(self, args, command_cls, length=1,

1784

clients=None, **cmd_attrs):

1163

1785

"""Assert that parsing ARGS should result in an instance of

1164

1786

COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""

1165

1787

options = self.parser.parse_args(args)

1166

1788

check_option_syntax(self.parser, options)

1167

1789

commands = commands_from_options(options)

1168

self.assertEqual(len(commands), 1)

1169

command = commands[0]

1170

self.assertIsInstance(command, command_cls)

1790

self.assertEqual(length, len(commands))

1791

for command in commands:

1792

if isinstance(command, command_cls):

1793

break

1794

else:

1795

self.assertIsInstance(command, command_cls)

1796

if clients is not None:

1797

self.assertEqual(clients, options.client)

1171

1798

for key, value in cmd_attrs.items():

1172

self.assertEqual(getattr(command, key), value)

1799

self.assertEqual(value, getattr(command, key))

1800

1801

def assert_commands_from_args(self, args, commands, clients=None):

1802

for cmd in commands:

1803

self.assert_command_from_args(args, cmd,

1804

length=len(commands),

1805

clients=clients)

1173

1806

1174

1807

def test_is_enabled_short(self):

1175

self.assert_command_from_args(["-V", "foo"],

1808

self.assert_command_from_args(["-V", "client"],

1176

1809

command.IsEnabled)

1177

1810

1178

1811

def test_approve(self):

1179

self.assert_command_from_args(["--approve", "foo"],

1812

self.assert_command_from_args(["--approve", "client"],

1180

1813

command.Approve)

1181

1814

1182

1815

def test_approve_short(self):

1183

self.assert_command_from_args(["-A", "foo"], command.Approve)

1816

self.assert_command_from_args(["-A", "client"],

1817

command.Approve)

1184

1818

1185

1819

def test_deny(self):

1186

self.assert_command_from_args(["--deny", "foo"], command.Deny)

1820

self.assert_command_from_args(["--deny", "client"],

1821

command.Deny)

1187

1822

1188

1823

def test_deny_short(self):

1189

self.assert_command_from_args(["-D", "foo"], command.Deny)

1824

self.assert_command_from_args(["-D", "client"], command.Deny)

1190

1825

1191

1826

def test_remove(self):

1192

self.assert_command_from_args(["--remove", "foo"],

1827

self.assert_command_from_args(["--remove", "client"],

1193

1828

command.Remove)

1194

1829

1195

1830

def test_deny_before_remove(self):

1196

1831

options = self.parser.parse_args(["--deny", "--remove",

1197

"foo"])

1832

"client"])

1198

1833

check_option_syntax(self.parser, options)

1199

1834

commands = commands_from_options(options)

1200

self.assertEqual(len(commands), 2)

1835

self.assertEqual(2, len(commands))

1201

1836

self.assertIsInstance(commands[0], command.Deny)

1202

1837

self.assertIsInstance(commands[1], command.Remove)

1203

1838

1206

1841

"--all"])

1207

1842

check_option_syntax(self.parser, options)

1208

1843

commands = commands_from_options(options)

1209

self.assertEqual(len(commands), 2)

1844

self.assertEqual(2, len(commands))

1210

1845

self.assertIsInstance(commands[0], command.Deny)

1211

1846

self.assertIsInstance(commands[1], command.Remove)

1212

1847

1213

1848

def test_remove_short(self):

1214

self.assert_command_from_args(["-r", "foo"], command.Remove)

1849

self.assert_command_from_args(["-r", "client"],

1850

command.Remove)

1215

1851

1216

1852

def test_dump_json(self):

1217

1853

self.assert_command_from_args(["--dump-json"],

1218

1854

command.DumpJSON)

1219

1855

1220

1856

def test_enable(self):

1221

self.assert_command_from_args(["--enable", "foo"],

1857

self.assert_command_from_args(["--enable", "client"],

1222

1858

command.Enable)

1223

1859

1224

1860

def test_enable_short(self):

1225

self.assert_command_from_args(["-e", "foo"], command.Enable)

1861

self.assert_command_from_args(["-e", "client"],

1862

command.Enable)

1226

1863

1227

1864

def test_disable(self):

1228

self.assert_command_from_args(["--disable", "foo"],

1865

self.assert_command_from_args(["--disable", "client"],

1229

1866

command.Disable)

1230

1867

1231

1868

def test_disable_short(self):

1232

self.assert_command_from_args(["-d", "foo"], command.Disable)

1869

self.assert_command_from_args(["-d", "client"],

1870

command.Disable)

1233

1871

1234

1872

def test_bump_timeout(self):

1235

self.assert_command_from_args(["--bump-timeout", "foo"],

1873

self.assert_command_from_args(["--bump-timeout", "client"],

1236

1874

command.BumpTimeout)

1237

1875

1238

1876

def test_bump_timeout_short(self):

1239

self.assert_command_from_args(["-b", "foo"],

1877

self.assert_command_from_args(["-b", "client"],

1240

1878

command.BumpTimeout)

1241

1879

1242

1880

def test_start_checker(self):

1243

self.assert_command_from_args(["--start-checker", "foo"],

1881

self.assert_command_from_args(["--start-checker", "client"],

1244

1882

command.StartChecker)

1245

1883

1246

1884

def test_stop_checker(self):

1247

self.assert_command_from_args(["--stop-checker", "foo"],

1885

self.assert_command_from_args(["--stop-checker", "client"],

1248

1886

command.StopChecker)

1249

1887

1250

1888

def test_approve_by_default(self):

1251

self.assert_command_from_args(["--approve-by-default", "foo"],

1889

self.assert_command_from_args(["--approve-by-default",

1890

"client"],

1252

1891

command.ApproveByDefault)

1253

1892

1254

1893

def test_deny_by_default(self):

1255

self.assert_command_from_args(["--deny-by-default", "foo"],

1894

self.assert_command_from_args(["--deny-by-default", "client"],

1256

1895

command.DenyByDefault)

1257

1896

1258

1897

def test_checker(self):

1259

self.assert_command_from_args(["--checker", ":", "foo"],

1898

self.assert_command_from_args(["--checker", ":", "client"],

1260

1899

command.SetChecker,

1261

1900

value_to_set=":")

1262

1901

1263

1902

def test_checker_empty(self):

1264

self.assert_command_from_args(["--checker", "", "foo"],

1903

self.assert_command_from_args(["--checker", "", "client"],

1265

1904

command.SetChecker,

1266

1905

value_to_set="")

1267

1906

1268

1907

def test_checker_short(self):

1269

self.assert_command_from_args(["-c", ":", "foo"],

1908

self.assert_command_from_args(["-c", ":", "client"],

1270

1909

command.SetChecker,

1271

1910

value_to_set=":")

1272

1911

1273

1912

def test_host(self):

1274

self.assert_command_from_args(["--host", "foo.example.org",

1275

"foo"], command.SetHost,

1276

value_to_set="foo.example.org")

1913

self.assert_command_from_args(

1914

["--host", "client.example.org", "client"],

1915

command.SetHost, value_to_set="client.example.org")

1277

1916

1278

1917

def test_host_short(self):

1279

self.assert_command_from_args(["-H", "foo.example.org",

1280

"foo"], command.SetHost,

1281

value_to_set="foo.example.org")

1918

self.assert_command_from_args(

1919

["-H", "client.example.org", "client"], command.SetHost,

1920

value_to_set="client.example.org")

1282

1921

1283

1922

def test_secret_devnull(self):

1284

1923

self.assert_command_from_args(["--secret", os.path.devnull,

1285

"foo"], command.SetSecret,

1924

"client"], command.SetSecret,

1286

1925

value_to_set=b"")

1287

1926

1288

1927

def test_secret_tempfile(self):

1291

1930

f.write(value)

1292

1931

f.seek(0)

1293

1932

self.assert_command_from_args(["--secret", f.name,

1294

"foo"], command.SetSecret,

1933

"client"],

1934

command.SetSecret,

1295

1935

value_to_set=value)

1296

1936

1297

1937

def test_secret_devnull_short(self):

1298

self.assert_command_from_args(["-s", os.path.devnull, "foo"],

1299

command.SetSecret,

1938

self.assert_command_from_args(["-s", os.path.devnull,

1939

"client"], command.SetSecret,

1300

1940

value_to_set=b"")

1301

1941

1302

1942

def test_secret_tempfile_short(self):

1304

1944

value = b"secret\0xyzzy\nbar"

1305

1945

f.write(value)

1306

1946

f.seek(0)

1307

self.assert_command_from_args(["-s", f.name, "foo"],

1947

self.assert_command_from_args(["-s", f.name, "client"],

1308

1948

command.SetSecret,

1309

1949

value_to_set=value)

1310

1950

1311

1951

def test_timeout(self):

1312

self.assert_command_from_args(["--timeout", "PT5M", "foo"],

1952

self.assert_command_from_args(["--timeout", "PT5M", "client"],

1313

1953

command.SetTimeout,

1314

1954

value_to_set=300000)

1315

1955

1316

1956

def test_timeout_short(self):

1317

self.assert_command_from_args(["-t", "PT5M", "foo"],

1957

self.assert_command_from_args(["-t", "PT5M", "client"],

1318

1958

command.SetTimeout,

1319

1959

value_to_set=300000)

1320

1960

1321

1961

def test_extended_timeout(self):

1322

1962

self.assert_command_from_args(["--extended-timeout", "PT15M",

1323

"foo"],

1963

"client"],

1324

1964

command.SetExtendedTimeout,

1325

1965

value_to_set=900000)

1326

1966

1327

1967

def test_interval(self):

1328

self.assert_command_from_args(["--interval", "PT2M", "foo"],

1329

command.SetInterval,

1968

self.assert_command_from_args(["--interval", "PT2M",

1969

"client"], command.SetInterval,

1330

1970

value_to_set=120000)

1331

1971

1332

1972

def test_interval_short(self):

1333

self.assert_command_from_args(["-i", "PT2M", "foo"],

1973

self.assert_command_from_args(["-i", "PT2M", "client"],

1334

1974

command.SetInterval,

1335

1975

value_to_set=120000)

1336

1976

1337

1977

def test_approval_delay(self):

1338

1978

self.assert_command_from_args(["--approval-delay", "PT30S",

1339

"foo"],

1979

"client"],

1340

1980

command.SetApprovalDelay,

1341

1981

value_to_set=30000)

1342

1982

1343

1983

def test_approval_duration(self):

1344

1984

self.assert_command_from_args(["--approval-duration", "PT1S",

1345

"foo"],

1985

"client"],

1346

1986

command.SetApprovalDuration,

1347

1987

value_to_set=1000)

1348

1988

1360

2000

verbose=True)

1361

2001

1362

2002

2003

def test_manual_page_example_1(self):

2004

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

2005

command.PrintTable,

2006

clients=[],

2007

verbose=True)

2008

2009

def test_manual_page_example_2(self):

2010

self.assert_command_from_args(

2011

"--verbose foo1.example.org foo2.example.org".split(),

2012

command.PrintTable, clients=["foo1.example.org",

2013

"foo2.example.org"],

2014

verbose=True)

2015

2016

def test_manual_page_example_3(self):

2017

self.assert_command_from_args("--enable --all".split(),

2018

command.Enable,

2019

clients=[])

2020

2021

def test_manual_page_example_4(self):

2022

self.assert_commands_from_args(

2023

("--timeout=PT5M --interval=PT1M foo1.example.org"

2024

" foo2.example.org").split(),

2025

[command.SetTimeout, command.SetInterval],

2026

clients=["foo1.example.org", "foo2.example.org"])

2027

2028

def test_manual_page_example_5(self):

2029

self.assert_command_from_args("--approve --all".split(),

2030

command.Approve,

2031

clients=[])

2032

2033

1363

2034

class TestCommand(unittest.TestCase):

1364

2035

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

1365

2036

2037

class FakeMandosBus(dbus.MandosBus):

2038

def __init__(self, testcase):

2039

self.client_properties = {

2040

"Name": "foo",

2041

"KeyID": ("92ed150794387c03ce684574b1139a65"

2042

"94a34f895daaaf09fd8ea90a27cddb12"),

2043

"Secret": b"secret",

2044

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

2045

"Enabled": True,

2046

"Timeout": 300000,

2047

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

2048

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

2049

"Interval": 120000,

2050

"Fingerprint": ("778827225BA7DE539C5A"

2051

"7CFA59CFF7CDBD9A5920"),

2052

"CheckerRunning": False,

2053

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

2054

"ApprovalPending": False,

2055

"ApprovedByDefault": True,

2056

"LastApprovalRequest": "",

2057

"ApprovalDelay": 0,

2058

"ApprovalDuration": 1000,

2059

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

2060

"ExtendedTimeout": 900000,

2061

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

2062

"LastCheckerStatus": 0,

2063

}

2064

self.other_client_properties = {

2065

"Name": "barbar",

2066

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2067

"6ab612cff5ad227247e46c2b020f441c"),

2068

"Secret": b"secretbar",

2069

"Host": "192.0.2.3",

2070

"Enabled": True,

2071

"Timeout": 300000,

2072

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

2073

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

2074

"Interval": 120000,

2075

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2076

"F547B3A107558FCA3A27"),

2077

"CheckerRunning": True,

2078

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

2079

"ApprovalPending": False,

2080

"ApprovedByDefault": False,

2081

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

2082

"ApprovalDelay": 30000,

2083

"ApprovalDuration": 93785000,

2084

"Checker": ":",

2085

"ExtendedTimeout": 900000,

2086

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

2087

"LastCheckerStatus": -2,

2088

}

2089

self.clients = collections.OrderedDict(

2090

[

2091

("client_objectpath", self.client_properties),

2092

("other_client_objectpath",

2093

self.other_client_properties),

2094

])

2095

self.one_client = {"client_objectpath":

2096

self.client_properties}

2097

self.testcase = testcase

2098

self.calls = []

2099

2100

def call_method(self, methodname, busname, objectpath,

2101

interface, *args):

2102

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

2103

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

2104

interface, args))

2105

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

2106

if methodname == "Set":

2107

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

2108

interface, key, value = args

2109

self.testcase.assertEqual(

2110

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

2111

self.clients[objectpath][key] = value

2112

return

2113

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

2114

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

2115

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

2116

clientpath = args[0]

2117

del self.clients[clientpath]

2118

return

2119

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

2120

if methodname == "Approve":

2121

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

2122

return

2123

raise ValueError()

2124

1366

2125

def setUp(self):

1367

testcase = self

1368

class MockClient(object):

1369

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

1370

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

1371

self.attributes = attributes

1372

self.attributes["Name"] = name

1373

self.calls = []

1374

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

1375

testcase.assertEqual(interface, client_dbus_interface)

1376

testcase.assertEqual(dbus_interface,

1377

dbus.PROPERTIES_IFACE)

1378

self.attributes[propname] = value

1379

def Get(self, interface, propname, dbus_interface):

1380

testcase.assertEqual(interface, client_dbus_interface)

1381

testcase.assertEqual(dbus_interface,

1382

dbus.PROPERTIES_IFACE)

1383

return self.attributes[propname]

1384

def Approve(self, approve, dbus_interface):

1385

testcase.assertEqual(dbus_interface,

1386

client_dbus_interface)

1387

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

1388

dbus_interface)))

1389

self.client = MockClient(

1390

"foo",

1391

KeyID=("92ed150794387c03ce684574b1139a65"

1392

"94a34f895daaaf09fd8ea90a27cddb12"),

1393

Secret=b"secret",

1394

Host="foo.example.org",

1395

Enabled=dbus.Boolean(True),

1396

Timeout=300000,

1397

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

1398

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

1399

Interval=120000,

1400

Fingerprint=("778827225BA7DE539C5A"

1401

"7CFA59CFF7CDBD9A5920"),

1402

CheckerRunning=dbus.Boolean(False),

1403

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

1404

ApprovalPending=dbus.Boolean(False),

1405

ApprovedByDefault=dbus.Boolean(True),

1406

LastApprovalRequest="",

1407

ApprovalDelay=0,

1408

ApprovalDuration=1000,

1409

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

1410

ExtendedTimeout=900000,

1411

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

1412

LastCheckerStatus=0)

1413

self.other_client = MockClient(

1414

"barbar",

1415

KeyID=("0558568eedd67d622f5c83b35a115f79"

1416

"6ab612cff5ad227247e46c2b020f441c"),

1417

Secret=b"secretbar",

1418

Host="192.0.2.3",

1419

Enabled=dbus.Boolean(True),

1420

Timeout=300000,

1421

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

1422

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

1423

Interval=120000,

1424

Fingerprint=("3E393AEAEFB84C7E89E2"

1425

"F547B3A107558FCA3A27"),

1426

CheckerRunning=dbus.Boolean(True),

1427

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

1428

ApprovalPending=dbus.Boolean(False),

1429

ApprovedByDefault=dbus.Boolean(False),

1430

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

1431

ApprovalDelay=30000,

1432

ApprovalDuration=93785000,

1433

Checker=":",

1434

ExtendedTimeout=900000,

1435

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

1436

LastCheckerStatus=-2)

1437

self.clients = collections.OrderedDict(

1438

[

1439

("/clients/foo", self.client.attributes),

1440

("/clients/barbar", self.other_client.attributes),

1441

])

1442

self.one_client = {"/clients/foo": self.client.attributes}

1443

1444

@property

1445

def bus(self):

1446

class Bus(object):

1447

@staticmethod

1448

def get_object(client_bus_name, path):

1449

self.assertEqual(client_bus_name, dbus_busname)

1450

return {

1451

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

1452

# the Bus instance, since this is a static method!

1453

"/clients/foo": self.client,

1454

"/clients/barbar": self.other_client,

1455

}[path]

1456

return Bus()

2126

self.bus = self.FakeMandosBus(self)

1457

2127

1458

2128

1459

2129

class TestBaseCommands(TestCommand):

1460

2130

1461

def test_is_enabled(self):

1462

self.assertTrue(all(command.IsEnabled().is_enabled(client,

1463

properties)

1464

for client, properties

1465

in self.clients.items()))

1466

1467

def test_is_enabled_run_exits_successfully(self):

2131

def test_IsEnabled_exits_successfully(self):

1468

2132

with self.assertRaises(SystemExit) as e:

1469

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

2133

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

1470

2134

if e.exception.code is not None:

1471

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

2135

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

1472

2136

else:

1473

2137

self.assertIsNone(e.exception.code)

1474

2138

1475

def test_is_enabled_run_exits_with_failure(self):

1476

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

2139

def test_IsEnabled_exits_with_failure(self):

2140

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

1477

2141

with self.assertRaises(SystemExit) as e:

1478

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

2142

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

1479

2143

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

1480

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

2144

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

1481

2145

else:

1482

2146

self.assertIsNotNone(e.exception.code)

1483

2147

1484

def test_approve(self):

1485

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

1486

for clientpath in self.clients:

1487

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

1488

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

1489

client.calls)

1490

1491

def test_deny(self):

1492

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

1493

for clientpath in self.clients:

1494

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

1495

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

1496

client.calls)

1497

1498

def test_remove(self):

1499

class MockMandos(object):

1500

def __init__(self):

1501

self.calls = []

1502

def RemoveClient(self, dbus_path):

1503

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

1504

mandos = MockMandos()

1505

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

1506

self.assertEqual(len(mandos.calls), 2)

1507

for clientpath in self.clients:

1508

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

1509

mandos.calls)

2148

def test_Approve(self):

2149

busname = "se.recompile.Mandos"

2150

client_interface = "se.recompile.Mandos.Client"

2151

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

2152

for clientpath in self.bus.clients:

2153

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

2154

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

2155

2156

def test_Deny(self):

2157

busname = "se.recompile.Mandos"

2158

client_interface = "se.recompile.Mandos.Client"

2159

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

2160

for clientpath in self.bus.clients:

2161

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

2162

client_interface, (False,)),

2163

self.bus.calls)

2164

2165

def test_Remove(self):

2166

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,

2170

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

1510

2171

1511

2172

expected_json = {

1512

2173

"foo": {

1560

2221

}

1561

2222

1562

2223

def test_DumpJSON_normal(self):

1563

output = command.DumpJSON().output(self.clients.values())

1564

json_data = json.loads(output)

1565

self.assertDictEqual(json_data, self.expected_json)

2224

with self.capture_stdout_to_buffer() as buffer:

2225

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

2226

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

2227

self.assertDictEqual(self.expected_json, json_data)

2228

2229

@staticmethod

2230

@contextlib.contextmanager

2231

def capture_stdout_to_buffer():

2232

capture_buffer = io.StringIO()

2233

old_stdout = sys.stdout

2234

sys.stdout = capture_buffer

2235

try:

2236

yield capture_buffer

2237

finally:

2238

sys.stdout = old_stdout

1566

2239

1567

2240

def test_DumpJSON_one_client(self):

1568

output = command.DumpJSON().output(self.one_client.values())

1569

json_data = json.loads(output)

2241

with self.capture_stdout_to_buffer() as buffer:

2242

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

2243

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

1570

2244

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

1571

self.assertDictEqual(json_data, expected_json)

2245

self.assertDictEqual(expected_json, json_data)

1572

2246

1573

2247

def test_PrintTable_normal(self):

1574

output = command.PrintTable().output(self.clients.values())

2248

with self.capture_stdout_to_buffer() as buffer:

2249

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

1575

2250

expected_output = "\n".join((

1576

2251

"Name Enabled Timeout Last Successful Check",

1577

2252

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

1578

2253

"barbar Yes 00:05:00 2019-02-04T00:00:00 ",

1579

))

1580

self.assertEqual(output, expected_output)

2254

)) + "\n"

2255

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

1581

2256

1582

2257

def test_PrintTable_verbose(self):

1583

output = command.PrintTable(verbose=True).output(

1584

self.clients.values())

2258

with self.capture_stdout_to_buffer() as buffer:

2259

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

1585

2260

columns = (

1586

2261

(

1587

2262

"Name ",

1669

2344

)

1670

2345

)

1671

2346

num_lines = max(len(rows) for rows in columns)

1672

expected_output = "\n".join("".join(rows[line]

1673

for rows in columns)

1674

for line in range(num_lines))

1675

self.assertEqual(output, expected_output)

2347

expected_output = ("\n".join("".join(rows[line]

2348

for rows in columns)

2349

for line in range(num_lines))

2350

+ "\n")

2351

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

1676

2352

1677

2353

def test_PrintTable_one_client(self):

1678

output = command.PrintTable().output(self.one_client.values())

2354

with self.capture_stdout_to_buffer() as buffer:

2355

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

1679

2356

expected_output = "\n".join((

1680

2357

"Name Enabled Timeout Last Successful Check",

1681

2358

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

1682

))

1683

self.assertEqual(output, expected_output)

1684

1685

1686

class TestPropertyCmd(TestCommand):

1687

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

2359

)) + "\n"

2360

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

2361

2362

2363

class TestPropertySetterCmd(TestCommand):

2364

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

2365

1688

2366

def runTest(self):

1689

2367

if not hasattr(self, "command"):

1690

return

1691

values_to_get = getattr(self, "values_to_get",

1692

self.values_to_set)

1693

for value_to_set, value_to_get in zip(self.values_to_set,

1694

values_to_get):

1695

for clientpath in self.clients:

1696

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

1697

old_value = client.attributes[self.propname]

1698

self.assertNotIsInstance(old_value, self.Unique)

1699

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

1700

self.run_command(value_to_set, self.clients)

1701

for clientpath in self.clients:

1702

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

1703

value = client.attributes[self.propname]

1704

self.assertNotIsInstance(value, self.Unique)

1705

self.assertEqual(value, value_to_get)

1706

1707

class Unique(object):

1708

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

1709

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

1710

1711

def run_command(self, value, clients):

1712

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

1713

1714

1715

class TestEnableCmd(TestPropertyCmd):

2368

return # Abstract TestCase class

2369

2370

if hasattr(self, "values_to_set"):

2371

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

2372

values_to_get = getattr(self, "values_to_get",

2373

self.values_to_set)

2374

else:

2375

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

2376

values_to_get = self.values_to_get

2377

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

2378

for clientpath in self.bus.clients:

2379

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

2380

Unique())

2381

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

2382

for clientpath in self.bus.clients:

2383

value = (self.bus.clients[clientpath]

2384

[self.propname])

2385

self.assertNotIsInstance(value, Unique)

2386

self.assertEqual(value_to_get, value)

2387

2388

2389

class TestEnableCmd(TestPropertySetterCmd):

1716

2390

command = command.Enable

1717

2391

propname = "Enabled"

1718

values_to_set = [dbus.Boolean(True)]

1719

1720

1721

class TestDisableCmd(TestPropertyCmd):

2392

values_to_get = [True]

2393

2394

2395

class TestDisableCmd(TestPropertySetterCmd):

1722

2396

command = command.Disable

1723

2397

propname = "Enabled"

1724

values_to_set = [dbus.Boolean(False)]

1725

1726

1727

class TestBumpTimeoutCmd(TestPropertyCmd):

2398

values_to_get = [False]

2399

2400

2401

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1728

2402

command = command.BumpTimeout

1729

2403

propname = "LastCheckedOK"

1730

values_to_set = [""]

1731

1732

1733

class TestStartCheckerCmd(TestPropertyCmd):

2404

values_to_get = [""]

2405

2406

2407

class TestStartCheckerCmd(TestPropertySetterCmd):

1734

2408

command = command.StartChecker

1735

2409

propname = "CheckerRunning"

1736

values_to_set = [dbus.Boolean(True)]

1737

1738

1739

class TestStopCheckerCmd(TestPropertyCmd):

2410

values_to_get = [True]

2411

2412

2413

class TestStopCheckerCmd(TestPropertySetterCmd):

1740

2414

command = command.StopChecker

1741

2415

propname = "CheckerRunning"

1742

values_to_set = [dbus.Boolean(False)]

1743

1744

1745

class TestApproveByDefaultCmd(TestPropertyCmd):

2416

values_to_get = [False]

2417

2418

2419

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1746

2420

command = command.ApproveByDefault

1747

2421

propname = "ApprovedByDefault"

1748

values_to_set = [dbus.Boolean(True)]

1749

1750

1751

class TestDenyByDefaultCmd(TestPropertyCmd):

2422

values_to_get = [True]

2423

2424

2425

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1752

2426

command = command.DenyByDefault

1753

2427

propname = "ApprovedByDefault"

1754

values_to_set = [dbus.Boolean(False)]

1755

1756

1757

class TestPropertyValueCmd(TestPropertyCmd):

1758

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

1759

1760

def runTest(self):

1761

if type(self) is TestPropertyValueCmd:

1762

return

1763

return super(TestPropertyValueCmd, self).runTest()

1764

1765

def run_command(self, value, clients):

1766

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

1767

1768

1769

class TestSetCheckerCmd(TestPropertyValueCmd):

2428

values_to_get = [False]

2429

2430

2431

class TestSetCheckerCmd(TestPropertySetterCmd):

1770

2432

command = command.SetChecker

1771

2433

propname = "Checker"

1772

2434

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

1773

2435

1774

2436

1775

class TestSetHostCmd(TestPropertyValueCmd):

2437

class TestSetHostCmd(TestPropertySetterCmd):

1776

2438

command = command.SetHost

1777

2439

propname = "Host"

1778

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

1779

1780

1781

class TestSetSecretCmd(TestPropertyValueCmd):

2440

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

2441

2442

2443

class TestSetSecretCmd(TestPropertySetterCmd):

1782

2444

command = command.SetSecret

1783

2445

propname = "Secret"

1784

2446

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

1785

2447

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

1786

values_to_get = [b"", b"secret\0xyzzy\nbar"]

1787

1788

1789

class TestSetTimeoutCmd(TestPropertyValueCmd):

2448

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

2449

2450

2451

class TestSetTimeoutCmd(TestPropertySetterCmd):

1790

2452

command = command.SetTimeout

1791

2453

propname = "Timeout"

1792

2454

values_to_set = [datetime.timedelta(),

1794

2456

datetime.timedelta(seconds=1),

1795

2457

datetime.timedelta(weeks=1),

1796

2458

datetime.timedelta(weeks=52)]

1797

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1798

1799

1800

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

2459

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

2460

2461

2462

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

1801

2463

command = command.SetExtendedTimeout

1802

2464

propname = "ExtendedTimeout"

1803

2465

values_to_set = [datetime.timedelta(),

1805

2467

datetime.timedelta(seconds=1),

1806

2468

datetime.timedelta(weeks=1),

1807

2469

datetime.timedelta(weeks=52)]

1808

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1809

1810

1811

class TestSetIntervalCmd(TestPropertyValueCmd):

2470

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

2471

2472

2473

class TestSetIntervalCmd(TestPropertySetterCmd):

1812

2474

command = command.SetInterval

1813

2475

propname = "Interval"

1814

2476

values_to_set = [datetime.timedelta(),

1816

2478

datetime.timedelta(seconds=1),

1817

2479

datetime.timedelta(weeks=1),

1818

2480

datetime.timedelta(weeks=52)]

1819

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1820

1821

1822

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

2481

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

2482

2483

2484

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

1823

2485

command = command.SetApprovalDelay

1824

2486

propname = "ApprovalDelay"

1825

2487

values_to_set = [datetime.timedelta(),

1827

2489

datetime.timedelta(seconds=1),

1828

2490

datetime.timedelta(weeks=1),

1829

2491

datetime.timedelta(weeks=52)]

1830

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1831

1832

1833

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

2492

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

2493

2494

2495

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

1834

2496

command = command.SetApprovalDuration

1835

2497

propname = "ApprovalDuration"

1836

2498

values_to_set = [datetime.timedelta(),

1838

2500

datetime.timedelta(seconds=1),

1839

2501

datetime.timedelta(weeks=1),

1840

2502

datetime.timedelta(weeks=52)]

1841

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

2503

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

1842

2504

1843

2505

1844

2506

Older »