/mandos/trunk : revision 1074

To get this branch, use:

bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk

« back to all changes in this revision

Viewing changes to mandos-ctl

Committer: Teddy Hogeborn
Date: 2019-03-16 04:51:11 UTC
Revision ID: teddy@recompile.se-20190316045111-63jv8d18i2tcakuw

mandos-ctl: Add comment

* mandos-ctl
(Test_get_managed_objects.test_logs_and_exits_on_dbus_error): Add a
comment.

files modified:
TODO

intro.xml

mandos

mandos-ctl

plugin-runner.c

Show diffs side-by-side

added added

removed removed

mandos-ctl

import tempfile

import contextlib

try:

import pydbus

import gi

dbus_python = None

except ImportError:

import dbus as dbus_python

pydbus = None

class gi(object):

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

class repository(object):

class GLib(object):

class Error(Exception):

pass

import dbus

# Show warnings by default

if not sys.warnoptions:

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"

def main():

parser = argparse.ArgumentParser()

add_command_line_options(parser)

options = parser.parse_args()

check_option_syntax(parser, options)

clientnames = options.client

if options.debug:

log.setLevel(logging.DEBUG)

if pydbus is not None:

bus = pydbus_adapter.CachingBus(pydbus)

else:

bus = dbus_python_adapter.CachingBus(dbus_python)

100

101

try:

102

all_clients = bus.get_clients_and_properties()

103

except dbus.ConnectFailed as e:

104

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

105

sys.exit(1)

106

except dbus.Error as e:

107

log.critical(

108

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

109

sys.exit(1)

bus = dbus.SystemBus()

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

110

113

111

114

# Compile dict of (clientpath: properties) to process

112

115

if not clientnames:

122

125

log.critical("Client not found on server: %r", name)

123

126

sys.exit(1)

124

127

128

# Run all commands on clients

125

129

commands = commands_from_options(options)

126

127

130

for command in commands:

128

command.run(clients, bus)

131

command.run(clients, bus, mandos_serv)

129

132

130

133

131

134

def add_command_line_options(parser):

136

139

help="Select all clients")

137

140

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

138

141

help="Print all fields")

139

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

140

action="append_const", default=[],

141

const=command.DumpJSON(),

142

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

142

143

help="Dump client data in JSON format")

143

144

enable_disable = parser.add_mutually_exclusive_group()

144

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

145

action="append_const", default=[],

146

const=command.Enable(),

145

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

147

146

help="Enable client")

148

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

149

action="append_const", default=[],

150

const=command.Disable(),

147

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

148

action="store_true",

151

149

help="disable client")

152

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

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

150

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

155

151

help="Bump timeout for client")

156

152

start_stop_checker = parser.add_mutually_exclusive_group()

157

153

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

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

154

action="store_true",

161

155

help="Start checker for client")

162

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

163

action="append_const", default=[],

164

const=command.StopChecker(),

156

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

157

action="store_true",

165

158

help="Stop checker for client")

166

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

167

action="append_const", default=[],

168

const=command.IsEnabled(),

159

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

169

160

help="Check if client is enabled")

170

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

171

action="append_const", default=[],

172

const=command.Remove(),

161

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

173

162

help="Remove client")

174

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

175

action="append", default=[],

176

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

163

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

177

164

help="Set checker command for client")

178

parser.add_argument(

179

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

180

default=[], metavar="TIME",

181

type=command.SetTimeout.argparse(string_to_delta),

182

help="Set timeout for client")

183

parser.add_argument(

184

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

185

default=[], metavar="TIME",

186

type=command.SetExtendedTimeout.argparse(string_to_delta),

187

help="Set extended timeout for client")

188

parser.add_argument(

189

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

190

default=[], metavar="TIME",

191

type=command.SetInterval.argparse(string_to_delta),

192

help="Set checker interval for client")

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

193

171

approve_deny_default = parser.add_mutually_exclusive_group()

194

172

approve_deny_default.add_argument(

195

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

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

173

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

174

default=None, dest="approved_by_default",

198

175

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

199

176

approve_deny_default.add_argument(

200

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

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

177

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

178

dest="approved_by_default",

203

179

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

204

parser.add_argument(

205

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

206

default=[], metavar="TIME",

207

type=command.SetApprovalDelay.argparse(string_to_delta),

208

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

209

parser.add_argument(

210

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

211

default=[], metavar="TIME",

212

type=command.SetApprovalDuration.argparse(string_to_delta),

213

help="Set duration of one client approval")

214

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

215

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

216

type=command.SetHost,

217

help="Set host for client")

218

parser.add_argument(

219

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

220

default=[], metavar="FILENAME",

221

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

222

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

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

223

188

approve_deny = parser.add_mutually_exclusive_group()

224

189

approve_deny.add_argument(

225

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

226

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

190

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

227

191

help="Approve any current client request")

228

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

229

action="append_const", default=[],

230

const=command.Deny(),

192

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

231

193

help="Deny any current client request")

232

194

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

233

195

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

270

232

>>> rfc3339_duration_to_delta("")

271

233

Traceback (most recent call last):

272

234

...

273

ValueError: Invalid RFC 3339 duration: ""

235

ValueError: Invalid RFC 3339 duration: u''

274

236

>>> # Must start with "P":

275

237

>>> rfc3339_duration_to_delta("1D")

276

238

Traceback (most recent call last):

277

239

...

278

ValueError: Invalid RFC 3339 duration: "1D"

240

ValueError: Invalid RFC 3339 duration: u'1D'

279

241

>>> # Must use correct order

280

242

>>> rfc3339_duration_to_delta("PT1S2M")

281

243

Traceback (most recent call last):

282

244

...

283

ValueError: Invalid RFC 3339 duration: "PT1S2M"

245

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

284

246

>>> # Time needs time marker

285

247

>>> rfc3339_duration_to_delta("P1H2S")

286

248

Traceback (most recent call last):

287

249

...

288

ValueError: Invalid RFC 3339 duration: "P1H2S"

250

ValueError: Invalid RFC 3339 duration: u'P1H2S'

289

251

>>> # Weeks can not be combined with anything else

290

252

>>> rfc3339_duration_to_delta("P1D2W")

291

253

Traceback (most recent call last):

292

254

...

293

ValueError: Invalid RFC 3339 duration: "P1D2W"

255

ValueError: Invalid RFC 3339 duration: u'P1D2W'

294

256

>>> rfc3339_duration_to_delta("P2W2H")

295

257

Traceback (most recent call last):

296

258

...

297

ValueError: Invalid RFC 3339 duration: "P2W2H"

259

ValueError: Invalid RFC 3339 duration: u'P2W2H'

298

260

"""

299

261

300

262

# Parsing an RFC 3339 duration with regular expressions is not

371

333

break

372

334

else:

373

335

# No currently valid tokens were found

374

raise ValueError("Invalid RFC 3339 duration: \"{}\""

336

raise ValueError("Invalid RFC 3339 duration: {!r}"

375

337

.format(duration))

376

338

# End token found

377

339

return value

424

386

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

425

387

argparse"""

426

388

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)

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

450

408

451

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

409

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

452

410

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

453

if options.verbose and has_commands(options):

411

if options.verbose and has_actions(options):

454

412

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

455

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

456

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

457

415

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

458

if options.all and not has_commands(options):

416

if options.all and not has_actions(options):

459

417

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

460

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

461

and len(options.client) > 1):

418

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

462

419

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

463

if (len(options.commands) > 1

464

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

465

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

466

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

467

468

469

class dbus(object):

470

471

class SystemBus(object):

472

473

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

474

def get_managed_objects(self, busname, objectpath):

475

return self.call_method("GetManagedObjects", busname,

476

objectpath,

477

self.object_manager_iface)

478

479

properties_iface = "org.freedesktop.DBus.Properties"

480

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

481

value):

482

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

483

self.properties_iface, interface, key,

484

value)

485

486

487

class MandosBus(SystemBus):

488

busname_domain = "se.recompile"

489

busname = busname_domain + ".Mandos"

490

server_path = "/"

491

server_interface = busname_domain + ".Mandos"

492

client_interface = busname_domain + ".Mandos.Client"

493

del busname_domain

494

495

def get_clients_and_properties(self):

496

managed_objects = self.get_managed_objects(

497

self.busname, self.server_path)

498

return {objpath: properties[self.client_interface]

499

for objpath, properties in managed_objects.items()

500

if self.client_interface in properties}

501

502

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

503

return self.set_property(self.busname, objectpath,

504

self.client_interface, key,

505

value)

506

507

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

508

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

509

self.client_interface, *args)

510

511

def call_server_method(self, method, *args):

512

return self.call_method(method, self.busname,

513

self.server_path,

514

self.server_interface, *args)

515

516

class Error(Exception):

517

pass

518

519

class ConnectFailed(Error):

520

pass

521

522

523

class dbus_python_adapter(object):

524

525

class SystemBus(dbus.MandosBus):

526

"""Use dbus-python"""

527

528

def __init__(self, module=dbus_python):

529

self.dbus_python = module

530

self.bus = self.dbus_python.SystemBus()

531

532

@contextlib.contextmanager

533

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

534

try:

535

yield

536

except self.dbus_python.exceptions.DBusException as e:

537

# This does what "raise from" would do

538

exc = exception_class(*e.args)

539

exc.__cause__ = e

540

raise exc

541

542

def call_method(self, methodname, busname, objectpath,

543

interface, *args):

544

proxy_object = self.get_object(busname, objectpath)

545

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

546

interface, methodname,

547

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

548

method = getattr(proxy_object, methodname)

549

with self.convert_exception():

550

with dbus_python_adapter.SilenceLogger(

551

"dbus.proxies"):

552

value = method(*args, dbus_interface=interface)

553

return self.type_filter(value)

554

555

def get_object(self, busname, objectpath):

556

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

557

busname, objectpath)

558

with self.convert_exception(dbus.ConnectFailed):

559

return self.bus.get_object(busname, objectpath)

560

561

def type_filter(self, value):

562

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

563

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

564

return bool(value)

565

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

566

return str(value)

567

# Also recurse into dictionaries

568

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

569

return {self.type_filter(key):

570

self.type_filter(subval)

571

for key, subval in value.items()}

572

return value

573

574

575

class SilenceLogger(object):

576

"Simple context manager to silence a particular logger"

577

def __init__(self, loggername):

578

self.logger = logging.getLogger(loggername)

579

580

def __enter__(self):

581

self.logger.addFilter(self.nullfilter)

582

583

class NullFilter(logging.Filter):

584

def filter(self, record):

585

return False

586

587

nullfilter = NullFilter()

588

589

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

590

self.logger.removeFilter(self.nullfilter)

591

592

593

class CachingBus(SystemBus):

594

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

595

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

596

self.object_cache = {}

597

super(dbus_python_adapter.CachingBus,

598

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

599

def get_object(self, busname, objectpath):

600

try:

601

return self.object_cache[(busname, objectpath)]

602

except KeyError:

603

new_object = super(

604

dbus_python_adapter.CachingBus,

605

self).get_object(busname, objectpath)

606

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

607

return new_object

608

609

610

class pydbus_adapter(object):

611

class SystemBus(dbus.MandosBus):

612

def __init__(self, module=pydbus):

613

self.pydbus = module

614

self.bus = self.pydbus.SystemBus()

615

616

@contextlib.contextmanager

617

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

618

try:

619

yield

620

except gi.repository.GLib.Error as e:

621

# This does what "raise from" would do

622

exc = exception_class(*e.args)

623

exc.__cause__ = e

624

raise exc

625

626

def call_method(self, methodname, busname, objectpath,

627

interface, *args):

628

proxy_object = self.get(busname, objectpath)

629

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

630

interface, methodname,

631

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

632

method = getattr(proxy_object[interface], methodname)

633

with self.convert_exception():

634

return method(*args)

635

636

def get(self, busname, objectpath):

637

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

638

busname, objectpath)

639

with self.convert_exception(dbus.ConnectFailed):

640

if sys.version_info.major <= 2:

641

with warnings.catch_warnings():

642

warnings.filterwarnings(

643

"ignore", "", DeprecationWarning,

644

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

645

return self.bus.get(busname, objectpath)

646

else:

647

return self.bus.get(busname, objectpath)

648

649

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

650

value):

651

proxy_object = self.get(busname, objectpath)

652

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

653

objectpath, self.properties_iface, interface,

654

key, value)

655

setattr(proxy_object[interface], key, value)

656

657

class CachingBus(SystemBus):

658

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

659

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

660

self.object_cache = {}

661

super(pydbus_adapter.CachingBus,

662

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

663

def get(self, busname, objectpath):

664

try:

665

return self.object_cache[(busname, objectpath)]

666

except KeyError:

667

new_object = (super(pydbus_adapter.CachingBus, self)

668

.get(busname, objectpath))

669

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

670

return new_object

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)

671

473

672

474

673

475

def commands_from_options(options):

674

476

675

commands = list(options.commands)

676

677

def find_cmd(cmd, commands):

678

i = 0

679

for i, c in enumerate(commands):

680

if isinstance(c, cmd):

681

return i

682

return i+1

683

684

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

685

# to occur ahead of command.Remove.

686

index_of_remove = find_cmd(command.Remove, commands)

687

before_remove = commands[:index_of_remove]

688

after_remove = commands[index_of_remove:]

689

cleaned_after = []

690

for cmd in after_remove:

691

if isinstance(cmd, command.Deny):

692

before_remove.append(cmd)

477

commands = []

478

479

if options.is_enabled:

480

commands.append(IsEnabledCmd())

481

482

if options.approve:

483

commands.append(ApproveCmd())

484

485

if options.deny:

486

commands.append(DenyCmd())

487

488

if options.remove:

489

commands.append(RemoveCmd())

490

491

if options.dump_json:

492

commands.append(DumpJSONCmd())

493

494

if options.enable:

495

commands.append(EnableCmd())

496

497

if options.disable:

498

commands.append(DisableCmd())

499

500

if options.bump_timeout:

501

commands.append(BumpTimeoutCmd())

502

503

if options.start_checker:

504

commands.append(StartCheckerCmd())

505

506

if options.stop_checker:

507

commands.append(StopCheckerCmd())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(ApproveByDefaultCmd())

693

512

else:

694

cleaned_after.append(cmd)

695

if cleaned_after != after_remove:

696

commands = before_remove + cleaned_after

513

commands.append(DenyByDefaultCmd())

514

515

if options.checker is not None:

516

commands.append(SetCheckerCmd(options.checker))

517

518

if options.host is not None:

519

commands.append(SetHostCmd(options.host))

520

521

if options.secret is not None:

522

commands.append(SetSecretCmd(options.secret))

523

524

if options.timeout is not None:

525

commands.append(SetTimeoutCmd(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

SetExtendedTimeoutCmd(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(SetIntervalCmd(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(SetApprovalDelayCmd(options.approval_delay))

536

537

if options.approval_duration is not None:

538

commands.append(

539

SetApprovalDurationCmd(options.approval_duration))

697

540

698

541

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

699

542

# optionally verbosely

700

543

if not commands:

701

commands.append(command.PrintTable(verbose=options.verbose))

544

commands.append(PrintTableCmd(verbose=options.verbose))

702

545

703

546

return commands

704

547

705

548

706

class command(object):

707

"""A namespace for command classes"""

708

709

class Base(object):

710

"""Abstract base class for commands"""

711

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

712

"""Normal commands should implement run_on_one_client(),

713

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

714

override this run() method instead.

715

"""

716

self.bus = bus

717

for client, properties in clients.items():

718

self.run_on_one_client(client, properties)

719

720

721

class IsEnabled(Base):

722

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

723

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

724

if properties["Enabled"]:

725

sys.exit(0)

726

sys.exit(1)

727

728

729

class Approve(Base):

730

def run_on_one_client(self, client, properties):

731

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

732

733

734

class Deny(Base):

735

def run_on_one_client(self, client, properties):

736

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

737

738

739

class Remove(Base):

740

def run(self, clients, bus):

741

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

742

bus.call_server_method("RemoveClient", clientpath)

743

744

745

class Output(Base):

746

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

747

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

748

"Created", "Interval", "Host", "KeyID",

749

"Fingerprint", "CheckerRunning",

750

"LastEnabled", "ApprovalPending",

751

"ApprovedByDefault", "LastApprovalRequest",

752

"ApprovalDelay", "ApprovalDuration",

753

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

754

"LastCheckerStatus")

755

756

757

class DumpJSON(Output):

758

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

759

data = {properties["Name"]:

760

{key: properties[key]

761

for key in self.all_keywords}

762

for properties in clients.values()}

763

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

764

765

766

class PrintTable(Output):

767

def __init__(self, verbose=False):

768

self.verbose = verbose

769

770

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

771

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

772

"LastCheckedOK")

773

keywords = default_keywords

774

if self.verbose:

775

keywords = self.all_keywords

776

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

777

778

class TableOfClients(object):

779

tableheaders = {

780

"Name": "Name",

781

"Enabled": "Enabled",

782

"Timeout": "Timeout",

783

"LastCheckedOK": "Last Successful Check",

784

"LastApprovalRequest": "Last Approval Request",

785

"Created": "Created",

786

"Interval": "Interval",

787

"Host": "Host",

788

"Fingerprint": "Fingerprint",

789

"KeyID": "Key ID",

790

"CheckerRunning": "Check Is Running",

791

"LastEnabled": "Last Enabled",

792

"ApprovalPending": "Approval Is Pending",

793

"ApprovedByDefault": "Approved By Default",

794

"ApprovalDelay": "Approval Delay",

795

"ApprovalDuration": "Approval Duration",

796

"Checker": "Checker",

797

"ExtendedTimeout": "Extended Timeout",

798

"Expires": "Expires",

799

"LastCheckerStatus": "Last Checker Status",

800

}

801

802

def __init__(self, clients, keywords):

803

self.clients = clients

804

self.keywords = keywords

805

549

class Command(object):

550

"""Abstract class for commands"""

551

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

552

"""Normal commands should implement run_on_one_client(), but

553

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

554

can override this run() method instead."""

555

self.mandos = mandos

556

for clientpath, properties in clients.items():

557

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

558

dbus_busname, str(clientpath))

559

client = bus.get_object(dbus_busname, clientpath)

560

self.run_on_one_client(client, properties)

561

562

563

class IsEnabledCmd(Command):

564

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

565

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

566

if self.is_enabled(client, properties):

567

sys.exit(0)

568

sys.exit(1)

569

def is_enabled(self, client, properties):

570

return properties["Enabled"]

571

572

573

class ApproveCmd(Command):

574

def run_on_one_client(self, client, properties):

575

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

576

client.__dbus_object_path__, client_dbus_interface)

577

client.Approve(dbus.Boolean(True),

578

dbus_interface=client_dbus_interface)

579

580

581

class DenyCmd(Command):

582

def run_on_one_client(self, client, properties):

583

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

584

client.__dbus_object_path__, client_dbus_interface)

585

client.Approve(dbus.Boolean(False),

586

dbus_interface=client_dbus_interface)

587

588

589

class RemoveCmd(Command):

590

def run_on_one_client(self, client, properties):

591

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

592

server_dbus_path, server_dbus_interface,

593

str(client.__dbus_object_path__))

594

self.mandos.RemoveClient(client.__dbus_object_path__)

595

596

597

class OutputCmd(Command):

598

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

599

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

600

"Created", "Interval", "Host", "KeyID",

601

"Fingerprint", "CheckerRunning", "LastEnabled",

602

"ApprovalPending", "ApprovedByDefault",

603

"LastApprovalRequest", "ApprovalDelay",

604

"ApprovalDuration", "Checker", "ExtendedTimeout",

605

"Expires", "LastCheckerStatus")

606

607

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

608

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

609

610

def output(self, clients):

611

raise NotImplementedError()

612

613

614

class DumpJSONCmd(OutputCmd):

615

def output(self, clients):

616

data = {client["Name"]:

617

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

618

for key in self.all_keywords}

619

for client in clients}

620

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

621

622

@staticmethod

623

def dbus_boolean_to_bool(value):

624

if isinstance(value, dbus.Boolean):

625

value = bool(value)

626

return value

627

628

629

class PrintTableCmd(OutputCmd):

630

def __init__(self, verbose=False):

631

self.verbose = verbose

632

633

def output(self, clients):

634

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

635

"LastCheckedOK")

636

keywords = default_keywords

637

if self.verbose:

638

keywords = self.all_keywords

639

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

640

641

class TableOfClients(object):

642

tableheaders = {

643

"Name": "Name",

644

"Enabled": "Enabled",

645

"Timeout": "Timeout",

646

"LastCheckedOK": "Last Successful Check",

647

"LastApprovalRequest": "Last Approval Request",

648

"Created": "Created",

649

"Interval": "Interval",

650

"Host": "Host",

651

"Fingerprint": "Fingerprint",

652

"KeyID": "Key ID",

653

"CheckerRunning": "Check Is Running",

654

"LastEnabled": "Last Enabled",

655

"ApprovalPending": "Approval Is Pending",

656

"ApprovedByDefault": "Approved By Default",

657

"ApprovalDelay": "Approval Delay",

658

"ApprovalDuration": "Approval Duration",

659

"Checker": "Checker",

660

"ExtendedTimeout": "Extended Timeout",

661

"Expires": "Expires",

662

"LastCheckerStatus": "Last Checker Status",

663

}

664

665

def __init__(self, clients, keywords):

666

self.clients = clients

667

self.keywords = keywords

668

669

def __str__(self):

670

return "\n".join(self.rows())

671

672

if sys.version_info.major == 2:

673

__unicode__ = __str__

806

674

def __str__(self):

807

return "\n".join(self.rows())

808

809

if sys.version_info.major == 2:

810

__unicode__ = __str__

811

def __str__(self):

812

return str(self).encode(

813

locale.getpreferredencoding())

814

815

def rows(self):

816

format_string = self.row_formatting_string()

817

rows = [self.header_line(format_string)]

818

rows.extend(self.client_line(client, format_string)

819

for client in self.clients)

820

return rows

821

822

def row_formatting_string(self):

823

"Format string used to format table rows"

824

return " ".join("{{{key}:{width}}}".format(

825

width=max(len(self.tableheaders[key]),

826

*(len(self.string_from_client(client,

827

key))

828

for client in self.clients)),

829

key=key)

830

for key in self.keywords)

831

832

def string_from_client(self, client, key):

833

return self.valuetostring(client[key], key)

834

835

@classmethod

836

def valuetostring(cls, value, keyword):

837

if isinstance(value, bool):

838

return "Yes" if value else "No"

839

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

840

"ApprovalDuration", "ExtendedTimeout"):

841

return cls.milliseconds_to_string(value)

842

return str(value)

843

844

def header_line(self, format_string):

845

return format_string.format(**self.tableheaders)

846

847

def client_line(self, client, format_string):

848

return format_string.format(

849

**{key: self.string_from_client(client, key)

850

for key in self.keywords})

851

852

@staticmethod

853

def milliseconds_to_string(ms):

854

td = datetime.timedelta(0, 0, 0, ms)

855

return ("{days}{hours:02}:{minutes:02}:{seconds:02}"

856

.format(days="{}T".format(td.days)

857

if td.days else "",

858

hours=td.seconds // 3600,

859

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

860

seconds=td.seconds % 60))

861

862

863

class PropertySetter(Base):

864

"Abstract class for Actions for setting one client property"

865

866

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

867

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

868

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

869

self.value_to_set)

870

871

@property

872

def propname(self):

873

raise NotImplementedError()

874

875

876

class Enable(PropertySetter):

877

propname = "Enabled"

878

value_to_set = True

879

880

881

class Disable(PropertySetter):

882

propname = "Enabled"

883

value_to_set = False

884

885

886

class BumpTimeout(PropertySetter):

887

propname = "LastCheckedOK"

888

value_to_set = ""

889

890

891

class StartChecker(PropertySetter):

892

propname = "CheckerRunning"

893

value_to_set = True

894

895

896

class StopChecker(PropertySetter):

897

propname = "CheckerRunning"

898

value_to_set = False

899

900

901

class ApproveByDefault(PropertySetter):

902

propname = "ApprovedByDefault"

903

value_to_set = True

904

905

906

class DenyByDefault(PropertySetter):

907

propname = "ApprovedByDefault"

908

value_to_set = False

909

910

911

class PropertySetterValue(PropertySetter):

912

"""Abstract class for PropertySetter recieving a value as

913

constructor argument instead of a class attribute."""

914

def __init__(self, value):

915

self.value_to_set = value

675

return str(self).encode(locale.getpreferredencoding())

676

677

def rows(self):

678

format_string = self.row_formatting_string()

679

rows = [self.header_line(format_string)]

680

rows.extend(self.client_line(client, format_string)

681

for client in self.clients)

682

return rows

683

684

def row_formatting_string(self):

685

"Format string used to format table rows"

686

return " ".join("{{{key}:{width}}}".format(

687

width=max(len(self.tableheaders[key]),

688

*(len(self.string_from_client(client, key))

689

for client in self.clients)),

690

key=key)

691

for key in self.keywords)

692

693

def string_from_client(self, client, key):

694

return self.valuetostring(client[key], key)

916

695

917

696

@classmethod

918

def argparse(cls, argtype):

919

def cmdtype(arg):

920

return cls(argtype(arg))

921

return cmdtype

922

923

class SetChecker(PropertySetterValue):

924

propname = "Checker"

925

926

927

class SetHost(PropertySetterValue):

928

propname = "Host"

929

930

931

class SetSecret(PropertySetterValue):

932

propname = "Secret"

933

934

@property

935

def value_to_set(self):

936

return self._vts

937

938

@value_to_set.setter

939

def value_to_set(self, value):

940

"""When setting, read data from supplied file object"""

941

self._vts = value.read()

942

value.close()

943

944

945

class PropertySetterValueMilliseconds(PropertySetterValue):

946

"""Abstract class for PropertySetterValue taking a value

947

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

948

milliseconds."""

949

950

@property

951

def value_to_set(self):

952

return self._vts

953

954

@value_to_set.setter

955

def value_to_set(self, value):

956

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

957

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

958

959

960

class SetTimeout(PropertySetterValueMilliseconds):

961

propname = "Timeout"

962

963

964

class SetExtendedTimeout(PropertySetterValueMilliseconds):

965

propname = "ExtendedTimeout"

966

967

968

class SetInterval(PropertySetterValueMilliseconds):

969

propname = "Interval"

970

971

972

class SetApprovalDelay(PropertySetterValueMilliseconds):

973

propname = "ApprovalDelay"

974

975

976

class SetApprovalDuration(PropertySetterValueMilliseconds):

977

propname = "ApprovalDuration"

697

def valuetostring(cls, value, keyword):

698

if isinstance(value, dbus.Boolean):

699

return "Yes" if value else "No"

700

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

701

"ApprovalDuration", "ExtendedTimeout"):

702

return cls.milliseconds_to_string(value)

703

return str(value)

704

705

def header_line(self, format_string):

706

return format_string.format(**self.tableheaders)

707

708

def client_line(self, client, format_string):

709

return format_string.format(

710

**{key: self.string_from_client(client, key)

711

for key in self.keywords})

712

713

@staticmethod

714

def milliseconds_to_string(ms):

715

td = datetime.timedelta(0, 0, 0, ms)

716

return ("{days}{hours:02}:{minutes:02}:{seconds:02}"

717

.format(days="{}T".format(td.days)

718

if td.days else "",

719

hours=td.seconds // 3600,

720

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

721

seconds=td.seconds % 60))

722

723

724

class PropertyCmd(Command):

725

"""Abstract class for Actions for setting one client property"""

726

727

def run_on_one_client(self, client, properties):

728

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

729

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

730

client.__dbus_object_path__,

731

dbus.PROPERTIES_IFACE, client_dbus_interface,

732

self.propname, self.value_to_set

733

if not isinstance(self.value_to_set, dbus.Boolean)

734

else bool(self.value_to_set))

735

client.Set(client_dbus_interface, self.propname,

736

self.value_to_set,

737

dbus_interface=dbus.PROPERTIES_IFACE)

738

739

@property

740

def propname(self):

741

raise NotImplementedError()

742

743

744

class EnableCmd(PropertyCmd):

745

propname = "Enabled"

746

value_to_set = dbus.Boolean(True)

747

748

749

class DisableCmd(PropertyCmd):

750

propname = "Enabled"

751

value_to_set = dbus.Boolean(False)

752

753

754

class BumpTimeoutCmd(PropertyCmd):

755

propname = "LastCheckedOK"

756

value_to_set = ""

757

758

759

class StartCheckerCmd(PropertyCmd):

760

propname = "CheckerRunning"

761

value_to_set = dbus.Boolean(True)

762

763

764

class StopCheckerCmd(PropertyCmd):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(False)

767

768

769

class ApproveByDefaultCmd(PropertyCmd):

770

propname = "ApprovedByDefault"

771

value_to_set = dbus.Boolean(True)

772

773

774

class DenyByDefaultCmd(PropertyCmd):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(False)

777

778

779

class PropertyValueCmd(PropertyCmd):

780

"""Abstract class for PropertyCmd recieving a value as argument"""

781

def __init__(self, value):

782

self.value_to_set = value

783

784

785

class SetCheckerCmd(PropertyValueCmd):

786

propname = "Checker"

787

788

789

class SetHostCmd(PropertyValueCmd):

790

propname = "Host"

791

792

793

class SetSecretCmd(PropertyValueCmd):

794

propname = "Secret"

795

796

@property

797

def value_to_set(self):

798

return self._vts

799

800

@value_to_set.setter

801

def value_to_set(self, value):

802

"""When setting, read data from supplied file object"""

803

self._vts = value.read()

804

value.close()

805

806

807

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

808

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

809

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

810

811

@property

812

def value_to_set(self):

813

return self._vts

814

815

@value_to_set.setter

816

def value_to_set(self, value):

817

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

818

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

819

820

821

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

822

propname = "Timeout"

823

824

825

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

826

propname = "ExtendedTimeout"

827

828

829

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

830

propname = "Interval"

831

832

833

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

834

propname = "ApprovalDelay"

835

836

837

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

838

propname = "ApprovalDuration"

978

839

979

840

980

841

1011

872

("records",

1012

873

"output"))

1013

874

1014

1015

class Unique(object):

1016

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

1017

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

1018

1019

1020

875

class Test_string_to_delta(TestCaseWithAssertLogs):

1021

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

1022

# rfc3339_duration_to_delta() already has more comprehensive

1023

# tests, which are run by doctest.

1024

1025

def test_rfc3339_zero_seconds(self):

1026

self.assertEqual(datetime.timedelta(),

1027

string_to_delta("PT0S"))

1028

1029

def test_rfc3339_zero_days(self):

1030

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

1031

1032

def test_rfc3339_one_second(self):

1033

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

1034

string_to_delta("PT1S"))

1035

1036

def test_rfc3339_two_hours(self):

1037

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

1038

string_to_delta("PT2H"))

876

def test_handles_basic_rfc3339(self):

877

self.assertEqual(string_to_delta("PT0S"),

878

datetime.timedelta())

879

self.assertEqual(string_to_delta("P0D"),

880

datetime.timedelta())

881

self.assertEqual(string_to_delta("PT1S"),

882

datetime.timedelta(0, 1))

883

self.assertEqual(string_to_delta("PT2H"),

884

datetime.timedelta(0, 7200))

1039

885

1040

886

def test_falls_back_to_pre_1_6_1_with_warning(self):

1041

887

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

1042

888

value = string_to_delta("2h")

1043

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

889

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

1044

890

1045

891

1046

892

class Test_check_option_syntax(unittest.TestCase):

1050

896

1051

897

def test_actions_requires_client_or_all(self):

1052

898

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

1053

args = self.actionargs(action, value)

899

options = self.parser.parse_args()

900

setattr(options, action, value)

1054

901

with self.assertParseError():

1055

self.parse_args(args)

902

self.check_option_syntax(options)

1056

903

1057

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

904

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

1058

905

# check_option_syntax()

1059

906

actions = {

1060

"--enable": None,

1061

"--disable": None,

1062

"--bump-timeout": None,

1063

"--start-checker": None,

1064

"--stop-checker": None,

1065

"--is-enabled": None,

1066

"--remove": None,

1067

"--checker": "x",

1068

"--timeout": "PT0S",

1069

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

1070

"--interval": "PT0S",

1071

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

1072

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

1073

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

1074

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

1075

"--host": "hostname",

1076

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

1077

"--approve": None,

1078

"--deny": None,

907

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

908

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

909

# never used

910

"enable": True,

911

"disable": True,

912

"bump_timeout": True,

913

"start_checker": True,

914

"stop_checker": True,

915

"is_enabled": True,

916

"remove": True,

917

"checker": "x",

918

"timeout": datetime.timedelta(),

919

"extended_timeout": datetime.timedelta(),

920

"interval": datetime.timedelta(),

921

"approved_by_default": True,

922

"approval_delay": datetime.timedelta(),

923

"approval_duration": datetime.timedelta(),

924

"host": "x",

925

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

926

"approve": True,

927

"deny": True,

1079

928

}

1080

929

1081

@staticmethod

1082

def actionargs(action, value, *args):

1083

if value is not None:

1084

return [action, value] + list(args)

1085

else:

1086

return [action] + list(args)

1087

1088

930

@contextlib.contextmanager

1089

931

def assertParseError(self):

1090

932

with self.assertRaises(SystemExit) as e:

1091

with self.redirect_stderr_to_devnull():

933

with self.temporarily_suppress_stderr():

1092

934

yield

1093

935

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

1094

936

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

1095

937

# /argparse.html#exiting-methods

1096

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

1097

1098

def parse_args(self, args):

1099

options = self.parser.parse_args(args)

1100

check_option_syntax(self.parser, options)

938

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

1101

939

1102

940

@staticmethod

1103

941

@contextlib.contextmanager

1104

def redirect_stderr_to_devnull():

1105

old_stderr = sys.stderr

1106

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

1107

sys.stderr = null

1108

try:

1109

yield

1110

finally:

1111

sys.stderr = old_stderr

942

def temporarily_suppress_stderr():

943

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

944

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

945

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

946

os.close(null)

947

try:

948

yield

949

finally:

950

# restore stderr

951

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

952

os.close(stderrcopy)

1112

953

1113

954

def check_option_syntax(self, options):

1114

955

check_option_syntax(self.parser, options)

1115

956

1116

def test_actions_all_conflicts_with_verbose(self):

1117

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

1118

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

1119

"--verbose")

1120

with self.assertParseError():

1121

self.parse_args(args)

1122

1123

def test_actions_with_client_conflicts_with_verbose(self):

1124

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

1125

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

1126

"client")

1127

with self.assertParseError():

1128

self.parse_args(args)

957

def test_actions_conflicts_with_verbose(self):

958

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

959

options = self.parser.parse_args()

960

setattr(options, action, value)

961

options.verbose = True

962

with self.assertParseError():

963

self.check_option_syntax(options)

1129

964

1130

965

def test_dump_json_conflicts_with_verbose(self):

1131

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

966

options = self.parser.parse_args()

967

options.dump_json = True

968

options.verbose = True

1132

969

with self.assertParseError():

1133

self.parse_args(args)

970

self.check_option_syntax(options)

1134

971

1135

972

def test_dump_json_conflicts_with_action(self):

1136

973

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

1137

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

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.dump_json = True

1138

977

with self.assertParseError():

1139

self.parse_args(args)

978

self.check_option_syntax(options)

1140

979

1141

980

def test_all_can_not_be_alone(self):

1142

args = ["--all"]

981

options = self.parser.parse_args()

982

options.all = True

1143

983

with self.assertParseError():

1144

self.parse_args(args)

984

self.check_option_syntax(options)

1145

985

1146

986

def test_all_is_ok_with_any_action(self):

1147

987

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

1148

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

1149

self.parse_args(args)

1150

1151

def test_any_action_is_ok_with_one_client(self):

1152

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

1153

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

1154

self.parse_args(args)

1155

1156

def test_one_client_with_all_actions_except_is_enabled(self):

1157

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

1158

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

1159

continue

1160

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

1161

self.parse_args(args)

1162

1163

def test_two_clients_with_all_actions_except_is_enabled(self):

1164

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

1165

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

1166

continue

1167

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

1168

"client2")

1169

self.parse_args(args)

1170

1171

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1172

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

1173

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

1174

continue

1175

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

1176

"client2")

1177

self.parse_args(args)

988

options = self.parser.parse_args()

989

setattr(options, action, value)

990

options.all = True

991

self.check_option_syntax(options)

1178

992

1179

993

def test_is_enabled_fails_without_client(self):

1180

args = ["--is-enabled"]

994

options = self.parser.parse_args()

995

options.is_enabled = True

1181

996

with self.assertParseError():

1182

self.parse_args(args)

997

self.check_option_syntax(options)

998

999

def test_is_enabled_works_with_one_client(self):

1000

options = self.parser.parse_args()

1001

options.is_enabled = True

1002

options.client = ["foo"]

1003

self.check_option_syntax(options)

1183

1004

1184

1005

def test_is_enabled_fails_with_two_clients(self):

1185

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

1006

options = self.parser.parse_args()

1007

options.is_enabled = True

1008

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

1186

1009

with self.assertParseError():

1187

self.parse_args(args)

1010

self.check_option_syntax(options)

1188

1011

1189

1012

def test_remove_can_only_be_combined_with_action_deny(self):

1190

1013

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

1191

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

1014

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

1192

1015

continue

1193

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

1194

"--remove")

1016

options = self.parser.parse_args()

1017

setattr(options, action, value)

1018

options.all = True

1019

options.remove = True

1195

1020

with self.assertParseError():

1196

self.parse_args(args)

1197

1198

1199

class Test_dbus_exceptions(unittest.TestCase):

1200

1201

def test_dbus_ConnectFailed_is_Error(self):

1202

with self.assertRaises(dbus.Error):

1203

raise dbus.ConnectFailed()

1204

1205

1206

class Test_dbus_MandosBus(unittest.TestCase):

1207

1208

class MockMandosBus(dbus.MandosBus):

1209

def __init__(self):

1210

self._name = "se.recompile.Mandos"

1211

self._server_path = "/"

1212

self._server_interface = "se.recompile.Mandos"

1213

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

1214

self.calls = []

1215

self.call_method_return = Unique()

1216

1217

def call_method(self, methodname, busname, objectpath,

1218

interface, *args):

1219

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

1220

interface, args))

1221

return self.call_method_return

1222

1223

def setUp(self):

1224

self.bus = self.MockMandosBus()

1225

1226

def test_set_client_property(self):

1227

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

1228

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

1229

"org.freedesktop.DBus.Properties",

1230

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

1231

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

1232

1233

def test_call_client_method(self):

1234

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

1235

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

1236

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

1237

self.bus._client_interface, ())

1238

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

1239

1240

def test_call_client_method_with_args(self):

1241

args = (Unique(), Unique())

1242

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

1243

*args)

1244

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

1245

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

1246

self.bus._client_interface,

1247

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

1248

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

1249

1250

def test_get_clients_and_properties(self):

1251

managed_objects = {

1252

"objectpath": {

1253

self.bus._client_interface: {

1254

"key": "value",

1255

"bool": True,

1256

1257

"irrelevant_interface": {

1258

"key": "othervalue",

1259

"bool": False,

1260

1261

1262

"other_objectpath": {

1263

"other_irrelevant_interface": {

1264

"key": "value 3",

1265

"bool": None,

1266

1267

1268

}

1269

expected_clients_and_properties = {

1270

"objectpath": {

1271

"key": "value",

1272

"bool": True,

1273

}

1274

}

1275

self.bus.call_method_return = managed_objects

1276

ret = self.bus.get_clients_and_properties()

1277

self.assertDictEqual(expected_clients_and_properties, ret)

1278

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

1279

self.bus._server_path,

1280

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

1281

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

1282

1283

def test_call_server_method(self):

1284

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

1285

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

1286

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

1287

self.bus._server_path,

1288

self.bus._server_interface, ())

1289

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

1290

1291

def test_call_server_method_with_args(self):

1292

args = (Unique(), Unique())

1293

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

1294

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

1295

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

1296

self.bus._server_path,

1297

self.bus._server_interface,

1298

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

1299

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

1300

1301

1302

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1303

1304

def MockDBusPython_func(self, func):

1305

class mock_dbus_python(object):

1306

"""mock dbus-python module"""

1307

class exceptions(object):

1308

"""Pseudo-namespace"""

1309

class DBusException(Exception):

1310

pass

1311

class SystemBus(object):

1312

@staticmethod

1313

def get_object(busname, objectpath):

1314

DBusObject = collections.namedtuple(

1315

"DBusObject", ("methodname",))

1316

def method(*args, **kwargs):

1317

self.assertEqual({"dbus_interface":

1318

"interface"},

1319

kwargs)

1320

return func(*args)

1321

return DBusObject(methodname=method)

1322

class Boolean(object):

1323

def __init__(self, value):

1324

self.value = bool(value)

1325

def __bool__(self):

1326

return self.value

1327

if sys.version_info.major == 2:

1328

__nonzero__ = __bool__

1329

class ObjectPath(str):

1330

pass

1331

class Dictionary(dict):

1332

pass

1333

return mock_dbus_python

1334

1335

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

1336

interface, *args):

1337

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

1338

return bus.call_method(methodname, busname, objectpath,

1339

interface, *args)

1340

1341

def test_call_method_returns(self):

1342

expected_method_return = Unique()

1343

method_args = (Unique(), Unique())

1344

def func(*args):

1345

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

1346

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

1347

self.assertIs(marg, arg)

1348

return expected_method_return

1349

mock_dbus_python = self.MockDBusPython_func(func)

1350

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1351

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

1352

"objectpath", "interface",

1353

*method_args)

1354

self.assertIs(ret, expected_method_return)

1355

1356

def test_call_method_filters_bool_true(self):

1357

def func():

1358

return method_return

1359

mock_dbus_python = self.MockDBusPython_func(func)

1360

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1361

method_return = mock_dbus_python.Boolean(True)

1362

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

1363

"objectpath", "interface")

1364

self.assertTrue(ret)

1365

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1366

1367

def test_call_method_filters_bool_false(self):

1368

def func():

1369

return method_return

1370

mock_dbus_python = self.MockDBusPython_func(func)

1371

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1372

method_return = mock_dbus_python.Boolean(False)

1373

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

1374

"objectpath", "interface")

1375

self.assertFalse(ret)

1376

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1377

1378

def test_call_method_filters_objectpath(self):

1379

def func():

1380

return method_return

1381

mock_dbus_python = self.MockDBusPython_func(func)

1382

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1383

method_return = mock_dbus_python.ObjectPath("objectpath")

1384

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

1385

"objectpath", "interface")

1386

self.assertEqual("objectpath", ret)

1387

self.assertIsNot("objectpath", ret)

1388

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1389

1390

def test_call_method_filters_booleans_in_dict(self):

1391

def func():

1392

return method_return

1393

mock_dbus_python = self.MockDBusPython_func(func)

1394

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1395

method_return = mock_dbus_python.Dictionary(

1396

{mock_dbus_python.Boolean(True):

1397

mock_dbus_python.Boolean(False),

1398

mock_dbus_python.Boolean(False):

1399

mock_dbus_python.Boolean(True)})

1400

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

1401

"objectpath", "interface")

1402

expected_method_return = {True: False,

1403

False: True}

1404

self.assertEqual(expected_method_return, ret)

1405

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1406

1407

def test_call_method_filters_objectpaths_in_dict(self):

1408

def func():

1409

return method_return

1410

mock_dbus_python = self.MockDBusPython_func(func)

1411

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1412

method_return = mock_dbus_python.Dictionary(

1413

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1414

mock_dbus_python.ObjectPath("objectpath_value_1"),

1415

mock_dbus_python.ObjectPath("objectpath_key_2"):

1416

mock_dbus_python.ObjectPath("objectpath_value_2")})

1417

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

1418

"objectpath", "interface")

1419

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

1420

for key, value in

1421

method_return.items()}

1422

self.assertEqual(expected_method_return, ret)

1423

self.assertIsInstance(ret, dict)

1424

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1425

1426

def test_call_method_filters_dict_in_dict(self):

1427

def func():

1428

return method_return

1429

mock_dbus_python = self.MockDBusPython_func(func)

1430

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1431

method_return = mock_dbus_python.Dictionary(

1432

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

1433

"key12": "value12"}),

1434

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

1435

"key22": "value22"})})

1436

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

1437

"objectpath", "interface")

1438

expected_method_return = {

1439

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

1440

"key12": "value12"},

1441

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

1442

"key22": "value22"},

1443

}

1444

self.assertEqual(expected_method_return, ret)

1445

self.assertIsInstance(ret, dict)

1446

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1447

for key, value in ret.items():

1448

self.assertIsInstance(value, dict)

1449

self.assertEqual(expected_method_return[key], value)

1450

self.assertNotIsInstance(value,

1451

mock_dbus_python.Dictionary)

1452

1453

def test_call_method_filters_dict_three_deep(self):

1454

def func():

1455

return method_return

1456

mock_dbus_python = self.MockDBusPython_func(func)

1457

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1458

method_return = mock_dbus_python.Dictionary(

1459

{"key1":

1460

mock_dbus_python.Dictionary(

1461

{"key2":

1462

mock_dbus_python.Dictionary(

1463

{"key3":

1464

mock_dbus_python.Boolean(True),

1465

}),

1466

}),

1467

})

1468

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

1469

"objectpath", "interface")

1470

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

1471

self.assertEqual(expected_method_return, ret)

1472

self.assertIsInstance(ret, dict)

1473

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1474

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

1475

self.assertNotIsInstance(ret["key1"],

1476

mock_dbus_python.Dictionary)

1477

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

1478

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

1479

mock_dbus_python.Dictionary)

1480

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

1481

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

1482

mock_dbus_python.Boolean)

1483

1484

def test_call_method_handles_exception(self):

1021

self.check_option_syntax(options)

1022

1023

1024

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1025

def test_calls_and_returns_get_object_on_bus(self):

1026

class MockBus(object):

1027

called = False

1028

def get_object(mockbus_self, busname, dbus_path):

1029

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

1030

# this MockBus instance is in "mockbus_self".

1031

self.assertEqual(busname, dbus_busname)

1032

self.assertEqual(dbus_path, server_dbus_path)

1033

mockbus_self.called = True

1034

return mockbus_self

1035

1036

mockbus = get_mandos_dbus_object(bus=MockBus())

1037

self.assertIsInstance(mockbus, MockBus)

1038

self.assertTrue(mockbus.called)

1039

1040

def test_logs_and_exits_on_dbus_error(self):

1041

class MockBusFailing(object):

1042

def get_object(self, busname, dbus_path):

1043

raise dbus.exceptions.DBusException("Test")

1044

1045

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

1046

with self.assertRaises(SystemExit) as e:

1047

bus = get_mandos_dbus_object(bus=MockBusFailing())

1048

1049

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

1050

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

1051

else:

1052

self.assertIsNotNone(e.exception.code)

1053

1054

1055

class Test_get_managed_objects(TestCaseWithAssertLogs):

1056

def test_calls_and_returns_GetManagedObjects(self):

1057

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

1058

class MockObjectManager(object):

1059

def GetManagedObjects(self):

1060

return managed_objects

1061

retval = get_managed_objects(MockObjectManager())

1062

self.assertDictEqual(managed_objects, retval)

1063

1064

def test_logs_and_exits_on_dbus_error(self):

1485

1065

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

1486

1066

1487

def func():

1488

dbus_logger.error("Test")

1489

raise mock_dbus_python.exceptions.DBusException()

1490

1491

mock_dbus_python = self.MockDBusPython_func(func)

1492

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1067

class MockObjectManagerFailing(object):

1068

def GetManagedObjects(self):

1069

dbus_logger.error("Test")

1070

raise dbus.exceptions.DBusException("Test")

1493

1071

1494

1072

class CountingHandler(logging.Handler):

1495

1073

count = 0

1501

1079

dbus_logger.addHandler(counting_handler)

1502

1080

1503

1081

try:

1504

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

1505

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

1506

"objectpath", "interface")

1082

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

1083

with self.assertRaises(SystemExit) as e:

1084

get_managed_objects(MockObjectManagerFailing())

1507

1085

finally:

1508

1086

dbus_logger.removeFilter(counting_handler)

1509

1087

1510

self.assertNotIsInstance(e, dbus.ConnectFailed)

1511

1512

1088

# Make sure the dbus logger was suppressed

1513

self.assertEqual(0, counting_handler.count)

1514

1515

def test_get_object_converts_to_correct_exception(self):

1516

bus = dbus_python_adapter.SystemBus(

1517

self.fake_dbus_python_raises_exception_on_connect)

1518

with self.assertRaises(dbus.ConnectFailed):

1519

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

1520

"objectpath", "interface")

1521

1522

class fake_dbus_python_raises_exception_on_connect(object):

1523

"""fake dbus-python module"""

1524

class exceptions(object):

1525

"""Pseudo-namespace"""

1526

class DBusException(Exception):

1527

pass

1528

1529

@classmethod

1530

def SystemBus(cls):

1531

def get_object(busname, objectpath):

1532

raise cls.exceptions.DBusException()

1533

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

1534

return Bus(get_object=get_object)

1535

1536

1537

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1538

class mock_dbus_python(object):

1539

"""mock dbus-python modules"""

1540

class SystemBus(object):

1541

@staticmethod

1542

def get_object(busname, objectpath):

1543

return Unique()

1544

1545

def setUp(self):

1546

self.bus = dbus_python_adapter.CachingBus(

1547

self.mock_dbus_python)

1548

1549

def test_returns_distinct_objectpaths(self):

1550

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

1551

self.assertIsInstance(obj1, Unique)

1552

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

1553

self.assertIsInstance(obj2, Unique)

1554

self.assertIsNot(obj1, obj2)

1555

1556

def test_returns_distinct_busnames(self):

1557

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

1558

self.assertIsInstance(obj1, Unique)

1559

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

1560

self.assertIsInstance(obj2, Unique)

1561

self.assertIsNot(obj1, obj2)

1562

1563

def test_returns_distinct_both(self):

1564

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

1565

self.assertIsInstance(obj1, Unique)

1566

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

1567

self.assertIsInstance(obj2, Unique)

1568

self.assertIsNot(obj1, obj2)

1569

1570

def test_returns_same(self):

1571

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

1572

self.assertIsInstance(obj1, Unique)

1573

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

1574

self.assertIsInstance(obj2, Unique)

1575

self.assertIs(obj1, obj2)

1576

1577

def test_returns_same_old(self):

1578

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

1579

self.assertIsInstance(obj1, Unique)

1580

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

1581

self.assertIsInstance(obj2, Unique)

1582

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

1583

self.assertIsInstance(obj1b, Unique)

1584

self.assertIsNot(obj1, obj2)

1585

self.assertIsNot(obj2, obj1b)

1586

self.assertIs(obj1, obj1b)

1587

1588

1589

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1590

1591

def Stub_pydbus_func(self, func):

1592

class stub_pydbus(object):

1593

"""stub pydbus module"""

1594

class SystemBus(object):

1595

@staticmethod

1596

def get(busname, objectpath):

1597

DBusObject = collections.namedtuple(

1598

"DBusObject", ("methodname",))

1599

return {"interface":

1600

DBusObject(methodname=func)}

1601

return stub_pydbus

1602

1603

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

1604

interface, *args):

1605

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

1606

return bus.call_method(methodname, busname, objectpath,

1607

interface, *args)

1608

1609

def test_call_method_returns(self):

1610

expected_method_return = Unique()

1611

method_args = (Unique(), Unique())

1612

def func(*args):

1613

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

1614

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

1615

self.assertIs(marg, arg)

1616

return expected_method_return

1617

stub_pydbus = self.Stub_pydbus_func(func)

1618

bus = pydbus_adapter.SystemBus(stub_pydbus)

1619

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

1620

"objectpath", "interface",

1621

*method_args)

1622

self.assertIs(ret, expected_method_return)

1623

1624

def test_call_method_handles_exception(self):

1625

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

1626

1627

def func():

1628

raise gi.repository.GLib.Error()

1629

1630

stub_pydbus = self.Stub_pydbus_func(func)

1631

bus = pydbus_adapter.SystemBus(stub_pydbus)

1632

1633

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

1634

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

1635

"objectpath", "interface")

1636

1637

self.assertNotIsInstance(e, dbus.ConnectFailed)

1638

1639

def test_get_converts_to_correct_exception(self):

1640

bus = pydbus_adapter.SystemBus(

1641

self.fake_pydbus_raises_exception_on_connect)

1642

with self.assertRaises(dbus.ConnectFailed):

1643

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

1644

"objectpath", "interface")

1645

1646

class fake_pydbus_raises_exception_on_connect(object):

1647

"""fake dbus-python module"""

1648

@classmethod

1649

def SystemBus(cls):

1650

def get(busname, objectpath):

1651

raise gi.repository.GLib.Error()

1652

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

1653

return Bus(get=get)

1654

1655

def test_set_property_uses_setattr(self):

1656

class Object(object):

1657

pass

1658

obj = Object()

1659

class pydbus_spy(object):

1660

class SystemBus(object):

1661

@staticmethod

1662

def get(busname, objectpath):

1663

return {"interface": obj}

1664

bus = pydbus_adapter.SystemBus(pydbus_spy)

1665

value = Unique()

1666

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

1667

value)

1668

self.assertIs(value, obj.key)

1669

1670

def test_get_suppresses_xml_deprecation_warning(self):

1671

if sys.version_info.major >= 3:

1672

return

1673

class stub_pydbus_get(object):

1674

class SystemBus(object):

1675

@staticmethod

1676

def get(busname, objectpath):

1677

warnings.warn_explicit(

1678

"deprecated", DeprecationWarning,

1679

"xml.etree.ElementTree", 0)

1680

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1681

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

1682

warnings.simplefilter("always")

1683

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

1684

self.assertEqual(0, len(w))

1685

1686

1687

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1688

class stub_pydbus(object):

1689

"""stub pydbus module"""

1690

class SystemBus(object):

1691

@staticmethod

1692

def get(busname, objectpath):

1693

return Unique()

1694

1695

def setUp(self):

1696

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

1697

1698

def test_returns_distinct_objectpaths(self):

1699

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

1700

self.assertIsInstance(obj1, Unique)

1701

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

1702

self.assertIsInstance(obj2, Unique)

1703

self.assertIsNot(obj1, obj2)

1704

1705

def test_returns_distinct_busnames(self):

1706

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

1707

self.assertIsInstance(obj1, Unique)

1708

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

1709

self.assertIsInstance(obj2, Unique)

1710

self.assertIsNot(obj1, obj2)

1711

1712

def test_returns_distinct_both(self):

1713

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

1714

self.assertIsInstance(obj1, Unique)

1715

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

1716

self.assertIsInstance(obj2, Unique)

1717

self.assertIsNot(obj1, obj2)

1718

1719

def test_returns_same(self):

1720

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

1721

self.assertIsInstance(obj1, Unique)

1722

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

1723

self.assertIsInstance(obj2, Unique)

1724

self.assertIs(obj1, obj2)

1725

1726

def test_returns_same_old(self):

1727

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

1728

self.assertIsInstance(obj1, Unique)

1729

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

1730

self.assertIsInstance(obj2, Unique)

1731

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

1732

self.assertIsInstance(obj1b, Unique)

1733

self.assertIsNot(obj1, obj2)

1734

self.assertIsNot(obj2, obj1b)

1735

self.assertIs(obj1, obj1b)

1089

self.assertEqual(counting_handler.count, 0)

1090

1091

# Test that the dbus_logger still works

1092

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

1093

dbus_logger.error("Test")

1094

1095

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

1096

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

1097

else:

1098

self.assertIsNotNone(e.exception.code)

1736

1099

1737

1100

1738

1101

class Test_commands_from_options(unittest.TestCase):

1739

1740

1102

def setUp(self):

1741

1103

self.parser = argparse.ArgumentParser()

1742

1104

add_command_line_options(self.parser)

1743

1105

1744

1106

def test_is_enabled(self):

1745

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

1746

command.IsEnabled)

1107

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

1108

IsEnabledCmd)

1747

1109

1748

1110

def assert_command_from_args(self, args, command_cls,

1749

1111

**cmd_attrs):

1752

1114

options = self.parser.parse_args(args)

1753

1115

check_option_syntax(self.parser, options)

1754

1116

commands = commands_from_options(options)

1755

self.assertEqual(1, len(commands))

1117

self.assertEqual(len(commands), 1)

1756

1118

command = commands[0]

1757

1119

self.assertIsInstance(command, command_cls)

1758

1120

for key, value in cmd_attrs.items():

1759

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

1121

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

1760

1122

1761

1123

def test_is_enabled_short(self):

1762

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

1763

command.IsEnabled)

1124

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

1764

1125

1765

1126

def test_approve(self):

1766

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

1767

command.Approve)

1127

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

1128

ApproveCmd)

1768

1129

1769

1130

def test_approve_short(self):

1770

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

1771

command.Approve)

1131

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

1772

1132

1773

1133

def test_deny(self):

1774

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

1775

command.Deny)

1134

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

1776

1135

1777

1136

def test_deny_short(self):

1778

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

1137

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

1779

1138

1780

1139

def test_remove(self):

1781

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

1782

command.Remove)

1140

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

1141

RemoveCmd)

1783

1142

1784

1143

def test_deny_before_remove(self):

1785

1144

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

1786

"client"])

1145

"foo"])

1787

1146

check_option_syntax(self.parser, options)

1788

1147

commands = commands_from_options(options)

1789

self.assertEqual(2, len(commands))

1790

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

1791

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

1148

self.assertEqual(len(commands), 2)

1149

self.assertIsInstance(commands[0], DenyCmd)

1150

self.assertIsInstance(commands[1], RemoveCmd)

1792

1151

1793

1152

def test_deny_before_remove_reversed(self):

1794

1153

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

1795

1154

"--all"])

1796

1155

check_option_syntax(self.parser, options)

1797

1156

commands = commands_from_options(options)

1798

self.assertEqual(2, len(commands))

1799

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

1800

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

1157

self.assertEqual(len(commands), 2)

1158

self.assertIsInstance(commands[0], DenyCmd)

1159

self.assertIsInstance(commands[1], RemoveCmd)

1801

1160

1802

1161

def test_remove_short(self):

1803

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

1804

command.Remove)

1162

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

1805

1163

1806

1164

def test_dump_json(self):

1807

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

1808

command.DumpJSON)

1165

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

1809

1166

1810

1167

def test_enable(self):

1811

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

1812

command.Enable)

1168

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

1813

1169

1814

1170

def test_enable_short(self):

1815

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

1816

command.Enable)

1171

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

1817

1172

1818

1173

def test_disable(self):

1819

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

1820

command.Disable)

1174

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

1175

DisableCmd)

1821

1176

1822

1177

def test_disable_short(self):

1823

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

1824

command.Disable)

1178

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

1825

1179

1826

1180

def test_bump_timeout(self):

1827

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

1828

command.BumpTimeout)

1181

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

1182

BumpTimeoutCmd)

1829

1183

1830

1184

def test_bump_timeout_short(self):

1831

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

1832

command.BumpTimeout)

1185

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

1833

1186

1834

1187

def test_start_checker(self):

1835

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

1836

command.StartChecker)

1188

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

1189

StartCheckerCmd)

1837

1190

1838

1191

def test_stop_checker(self):

1839

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

1840

command.StopChecker)

1192

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

1193

StopCheckerCmd)

1841

1194

1842

1195

def test_approve_by_default(self):

1843

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

1844

"client"],

1845

command.ApproveByDefault)

1196

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

1197

ApproveByDefaultCmd)

1846

1198

1847

1199

def test_deny_by_default(self):

1848

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

1849

command.DenyByDefault)

1200

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

1201

DenyByDefaultCmd)

1850

1202

1851

1203

def test_checker(self):

1852

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

1853

command.SetChecker,

1854

value_to_set=":")

1204

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

1205

SetCheckerCmd, value_to_set=":")

1855

1206

1856

1207

def test_checker_empty(self):

1857

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

1858

command.SetChecker,

1859

value_to_set="")

1208

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

1209

SetCheckerCmd, value_to_set="")

1860

1210

1861

1211

def test_checker_short(self):

1862

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

1863

command.SetChecker,

1864

value_to_set=":")

1212

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

1213

SetCheckerCmd, value_to_set=":")

1865

1214

1866

1215

def test_host(self):

1867

self.assert_command_from_args(

1868

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

1869

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

1216

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

1217

"foo"], SetHostCmd,

1218

value_to_set="foo.example.org")

1870

1219

1871

1220

def test_host_short(self):

1872

self.assert_command_from_args(

1873

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

1874

value_to_set="client.example.org")

1221

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

1222

"foo"], SetHostCmd,

1223

value_to_set="foo.example.org")

1875

1224

1876

1225

def test_secret_devnull(self):

1877

1226

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

1878

"client"], command.SetSecret,

1227

"foo"], SetSecretCmd,

1879

1228

value_to_set=b"")

1880

1229

1881

1230

def test_secret_tempfile(self):

1884

1233

f.write(value)

1885

1234

f.seek(0)

1886

1235

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

1887

"client"],

1888

command.SetSecret,

1236

"foo"], SetSecretCmd,

1889

1237

value_to_set=value)

1890

1238

1891

1239

def test_secret_devnull_short(self):

1892

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

1893

"client"], command.SetSecret,

1894

value_to_set=b"")

1240

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

1241

SetSecretCmd, value_to_set=b"")

1895

1242

1896

1243

def test_secret_tempfile_short(self):

1897

1244

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1898

1245

value = b"secret\0xyzzy\nbar"

1899

1246

f.write(value)

1900

1247

f.seek(0)

1901

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

1902

command.SetSecret,

1248

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

1249

SetSecretCmd,

1903

1250

value_to_set=value)

1904

1251

1905

1252

def test_timeout(self):

1906

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

1907

command.SetTimeout,

1253

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

1254

SetTimeoutCmd,

1908

1255

value_to_set=300000)

1909

1256

1910

1257

def test_timeout_short(self):

1911

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

1912

command.SetTimeout,

1258

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

1259

SetTimeoutCmd,

1913

1260

value_to_set=300000)

1914

1261

1915

1262

def test_extended_timeout(self):

1916

1263

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

1917

"client"],

1918

command.SetExtendedTimeout,

1264

"foo"],

1265

SetExtendedTimeoutCmd,

1919

1266

value_to_set=900000)

1920

1267

1921

1268

def test_interval(self):

1922

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

1923

"client"], command.SetInterval,

1269

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

1270

SetIntervalCmd,

1924

1271

value_to_set=120000)

1925

1272

1926

1273

def test_interval_short(self):

1927

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

1928

command.SetInterval,

1274

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

1275

SetIntervalCmd,

1929

1276

value_to_set=120000)

1930

1277

1931

1278

def test_approval_delay(self):

1932

1279

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

1933

"client"],

1934

command.SetApprovalDelay,

1280

"foo"], SetApprovalDelayCmd,

1935

1281

value_to_set=30000)

1936

1282

1937

1283

def test_approval_duration(self):

1938

1284

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

1939

"client"],

1940

command.SetApprovalDuration,

1285

"foo"], SetApprovalDurationCmd,

1941

1286

value_to_set=1000)

1942

1287

1943

1288

def test_print_table(self):

1944

self.assert_command_from_args([], command.PrintTable,

1289

self.assert_command_from_args([], PrintTableCmd,

1945

1290

verbose=False)

1946

1291

1947

1292

def test_print_table_verbose(self):

1948

self.assert_command_from_args(["--verbose"],

1949

command.PrintTable,

1293

self.assert_command_from_args(["--verbose"], PrintTableCmd,

1950

1294

verbose=True)

1951

1295

1952

1296

def test_print_table_verbose_short(self):

1953

self.assert_command_from_args(["-v"], command.PrintTable,

1297

self.assert_command_from_args(["-v"], PrintTableCmd,

1954

1298

verbose=True)

1955

1299

1956

1300

1957

class TestCommand(unittest.TestCase):

1301

class TestCmd(unittest.TestCase):

1958

1302

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

1959

1303

1960

class FakeMandosBus(dbus.MandosBus):

1961

def __init__(self, testcase):

1962

self.client_properties = {

1304

def setUp(self):

1305

testcase = self

1306

class MockClient(object):

1307

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

1308

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

1309

self.attributes = attributes

1310

self.attributes["Name"] = name

1311

self.calls = []

1312

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

1313

testcase.assertEqual(interface, client_dbus_interface)

1314

testcase.assertEqual(dbus_interface,

1315

dbus.PROPERTIES_IFACE)

1316

self.attributes[propname] = value

1317

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

1318

testcase.assertEqual(interface, client_dbus_interface)

1319

testcase.assertEqual(dbus_interface,

1320

dbus.PROPERTIES_IFACE)

1321

return self.attributes[propname]

1322

def Approve(self, approve, dbus_interface):

1323

testcase.assertEqual(dbus_interface,

1324

client_dbus_interface)

1325

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

1326

dbus_interface)))

1327

self.client = MockClient(

1328

"foo",

1329

KeyID=("92ed150794387c03ce684574b1139a65"

1330

"94a34f895daaaf09fd8ea90a27cddb12"),

1331

Secret=b"secret",

1332

Host="foo.example.org",

1333

Enabled=dbus.Boolean(True),

1334

Timeout=300000,

1335

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

1336

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

1337

Interval=120000,

1338

Fingerprint=("778827225BA7DE539C5A"

1339

"7CFA59CFF7CDBD9A5920"),

1340

CheckerRunning=dbus.Boolean(False),

1341

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

1342

ApprovalPending=dbus.Boolean(False),

1343

ApprovedByDefault=dbus.Boolean(True),

1344

LastApprovalRequest="",

1345

ApprovalDelay=0,

1346

ApprovalDuration=1000,

1347

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

1348

ExtendedTimeout=900000,

1349

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

1350

LastCheckerStatus=0)

1351

self.other_client = MockClient(

1352

"barbar",

1353

KeyID=("0558568eedd67d622f5c83b35a115f79"

1354

"6ab612cff5ad227247e46c2b020f441c"),

1355

Secret=b"secretbar",

1356

Host="192.0.2.3",

1357

Enabled=dbus.Boolean(True),

1358

Timeout=300000,

1359

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

1360

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

1361

Interval=120000,

1362

Fingerprint=("3E393AEAEFB84C7E89E2"

1363

"F547B3A107558FCA3A27"),

1364

CheckerRunning=dbus.Boolean(True),

1365

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

1366

ApprovalPending=dbus.Boolean(False),

1367

ApprovedByDefault=dbus.Boolean(False),

1368

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

1369

ApprovalDelay=30000,

1370

ApprovalDuration=93785000,

1371

Checker=":",

1372

ExtendedTimeout=900000,

1373

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

1374

LastCheckerStatus=-2)

1375

self.clients = collections.OrderedDict(

1376

[

1377

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

1378

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

1379

])

1380

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

1381

1382

@property

1383

def bus(self):

1384

class Bus(object):

1385

@staticmethod

1386

def get_object(client_bus_name, path):

1387

self.assertEqual(client_bus_name, dbus_busname)

1388

return {

1389

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

1390

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

1391

"/clients/foo": self.client,

1392

"/clients/barbar": self.other_client,

1393

}[path]

1394

return Bus()

1395

1396

1397

class TestIsEnabledCmd(TestCmd):

1398

def test_is_enabled(self):

1399

self.assertTrue(all(IsEnabledCmd().is_enabled(client,

1400

properties)

1401

for client, properties

1402

in self.clients.items()))

1403

1404

def test_is_enabled_run_exits_successfully(self):

1405

with self.assertRaises(SystemExit) as e:

1406

IsEnabledCmd().run(self.one_client)

1407

if e.exception.code is not None:

1408

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

1409

else:

1410

self.assertIsNone(e.exception.code)

1411

1412

def test_is_enabled_run_exits_with_failure(self):

1413

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

1414

with self.assertRaises(SystemExit) as e:

1415

IsEnabledCmd().run(self.one_client)

1416

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

1417

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

1418

else:

1419

self.assertIsNotNone(e.exception.code)

1420

1421

1422

class TestApproveCmd(TestCmd):

1423

def test_approve(self):

1424

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

1425

for clientpath in self.clients:

1426

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

1427

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

1428

client.calls)

1429

1430

1431

class TestDenyCmd(TestCmd):

1432

def test_deny(self):

1433

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

1434

for clientpath in self.clients:

1435

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

1436

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

1437

client.calls)

1438

1439

1440

class TestRemoveCmd(TestCmd):

1441

def test_remove(self):

1442

class MockMandos(object):

1443

def __init__(self):

1444

self.calls = []

1445

def RemoveClient(self, dbus_path):

1446

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

1447

mandos = MockMandos()

1448

super(TestRemoveCmd, self).setUp()

1449

RemoveCmd().run(self.clients, self.bus, mandos)

1450

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

1451

for clientpath in self.clients:

1452

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

1453

mandos.calls)

1454

1455

1456

class TestDumpJSONCmd(TestCmd):

1457

def setUp(self):

1458

self.expected_json = {

1459

"foo": {

1963

1460

"Name": "foo",

1964

1461

"KeyID": ("92ed150794387c03ce684574b1139a65"

1965

1462

"94a34f895daaaf09fd8ea90a27cddb12"),

1966

"Secret": b"secret",

1967

1463

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

1968

1464

"Enabled": True,

1969

1465

"Timeout": 300000,

1983

1479

"ExtendedTimeout": 900000,

1984

1480

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

1985

1481

"LastCheckerStatus": 0,

1986

}

1987

self.other_client_properties = {

1482

1483

"barbar": {

1988

1484

"Name": "barbar",

1989

1485

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1990

1486

"6ab612cff5ad227247e46c2b020f441c"),

1991

"Secret": b"secretbar",

1992

1487

"Host": "192.0.2.3",

1993

1488

"Enabled": True,

1994

1489

"Timeout": 300000,

2008

1503

"ExtendedTimeout": 900000,

2009

1504

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

2010

1505

"LastCheckerStatus": -2,

2011

}

2012

self.clients = collections.OrderedDict(

2013

[

2014

("client_objectpath", self.client_properties),

2015

("other_client_objectpath",

2016

self.other_client_properties),

2017

])

2018

self.one_client = {"client_objectpath":

2019

self.client_properties}

2020

self.testcase = testcase

2021

self.calls = []

2022

2023

def call_method(self, methodname, busname, objectpath,

2024

interface, *args):

2025

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

2026

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

2027

interface, args))

2028

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

2029

if methodname == "Set":

2030

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

2031

interface, key, value = args

2032

self.testcase.assertEqual(

2033

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

2034

self.clients[objectpath][key] = value

2035

return

2036

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

2037

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

2038

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

2039

clientpath = args[0]

2040

del self.clients[clientpath]

2041

return

2042

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

2043

if methodname == "Approve":

2044

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

2045

return

2046

raise ValueError()

2047

2048

def setUp(self):

2049

self.bus = self.FakeMandosBus(self)

2050

2051

2052

class TestBaseCommands(TestCommand):

2053

2054

def test_IsEnabled_exits_successfully(self):

2055

with self.assertRaises(SystemExit) as e:

2056

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

2057

if e.exception.code is not None:

2058

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

2059

else:

2060

self.assertIsNone(e.exception.code)

2061

2062

def test_IsEnabled_exits_with_failure(self):

2063

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

2064

with self.assertRaises(SystemExit) as e:

2065

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

2066

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

2067

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

2068

else:

2069

self.assertIsNotNone(e.exception.code)

2070

2071

def test_Approve(self):

2072

busname = "se.recompile.Mandos"

2073

client_interface = "se.recompile.Mandos.Client"

2074

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

2075

for clientpath in self.bus.clients:

2076

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

2077

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

2078

2079

def test_Deny(self):

2080

busname = "se.recompile.Mandos"

2081

client_interface = "se.recompile.Mandos.Client"

2082

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

2083

for clientpath in self.bus.clients:

2084

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

2085

client_interface, (False,)),

2086

self.bus.calls)

2087

2088

def test_Remove(self):

2089

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

2090

for clientpath in self.bus.clients:

2091

self.assertIn(("RemoveClient", dbus_busname,

2092

dbus_server_path, dbus_server_interface,

2093

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

2094

2095

expected_json = {

2096

"foo": {

2097

"Name": "foo",

2098

"KeyID": ("92ed150794387c03ce684574b1139a65"

2099

"94a34f895daaaf09fd8ea90a27cddb12"),

2100

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

2101

"Enabled": True,

2102

"Timeout": 300000,

2103

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

2104

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

2105

"Interval": 120000,

2106

"Fingerprint": ("778827225BA7DE539C5A"

2107

"7CFA59CFF7CDBD9A5920"),

2108

"CheckerRunning": False,

2109

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

2110

"ApprovalPending": False,

2111

"ApprovedByDefault": True,

2112

"LastApprovalRequest": "",

2113

"ApprovalDelay": 0,

2114

"ApprovalDuration": 1000,

2115

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

2116

"ExtendedTimeout": 900000,

2117

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

2118

"LastCheckerStatus": 0,

2119

2120

"barbar": {

2121

"Name": "barbar",

2122

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2123

"6ab612cff5ad227247e46c2b020f441c"),

2124

"Host": "192.0.2.3",

2125

"Enabled": True,

2126

"Timeout": 300000,

2127

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

2128

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

2129

"Interval": 120000,

2130

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2131

"F547B3A107558FCA3A27"),

2132

"CheckerRunning": True,

2133

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

2134

"ApprovalPending": False,

2135

"ApprovedByDefault": False,

2136

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

2137

"ApprovalDelay": 30000,

2138

"ApprovalDuration": 93785000,

2139

"Checker": ":",

2140

"ExtendedTimeout": 900000,

2141

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

2142

"LastCheckerStatus": -2,

2143

2144

}

2145

2146

def test_DumpJSON_normal(self):

2147

with self.capture_stdout_to_buffer() as buffer:

2148

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

2149

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

2150

self.assertDictEqual(self.expected_json, json_data)

2151

2152

@staticmethod

2153

@contextlib.contextmanager

2154

def capture_stdout_to_buffer():

2155

capture_buffer = io.StringIO()

2156

old_stdout = sys.stdout

2157

sys.stdout = capture_buffer

2158

try:

2159

yield capture_buffer

2160

finally:

2161

sys.stdout = old_stdout

2162

2163

def test_DumpJSON_one_client(self):

2164

with self.capture_stdout_to_buffer() as buffer:

2165

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

2166

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

1506

1507

}

1508

return super(TestDumpJSONCmd, self).setUp()

1509

1510

def test_normal(self):

1511

output = DumpJSONCmd().output(self.clients.values())

1512

json_data = json.loads(output)

1513

self.assertDictEqual(json_data, self.expected_json)

1514

1515

def test_one_client(self):

1516

output = DumpJSONCmd().output(self.one_client.values())

1517

json_data = json.loads(output)

2167

1518

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

2168

self.assertDictEqual(expected_json, json_data)

2169

2170

def test_PrintTable_normal(self):

2171

with self.capture_stdout_to_buffer() as buffer:

2172

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

1519

self.assertDictEqual(json_data, expected_json)

1520

1521

1522

class TestPrintTableCmd(TestCmd):

1523

def test_normal(self):

1524

output = PrintTableCmd().output(self.clients.values())

2173

1525

expected_output = "\n".join((

2174

1526

"Name Enabled Timeout Last Successful Check",

2175

1527

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

2176

1528

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

2177

)) + "\n"

2178

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

1529

))

1530

self.assertEqual(output, expected_output)

2179

1531

2180

def test_PrintTable_verbose(self):

2181

with self.capture_stdout_to_buffer() as buffer:

2182

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

1532

def test_verbose(self):

1533

output = PrintTableCmd(verbose=True).output(

1534

self.clients.values())

2183

1535

columns = (

2184

1536

(

2185

1537

"Name ",

2267

1619

)

2268

1620

)

2269

1621

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

2270

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

2271

for rows in columns)

2272

for line in range(num_lines))

2273

+ "\n")

2274

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

1622

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

1623

for rows in columns)

1624

for line in range(num_lines))

1625

self.assertEqual(output, expected_output)

2275

1626

2276

def test_PrintTable_one_client(self):

2277

with self.capture_stdout_to_buffer() as buffer:

2278

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

1627

def test_one_client(self):

1628

output = PrintTableCmd().output(self.one_client.values())

2279

1629

expected_output = "\n".join((

2280

1630

"Name Enabled Timeout Last Successful Check",

2281

1631

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

2282

)) + "\n"

2283

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

2284

2285

2286

class TestPropertySetterCmd(TestCommand):

2287

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

2288

1632

))

1633

self.assertEqual(output, expected_output)

1634

1635

1636

class TestPropertyCmd(TestCmd):

1637

"""Abstract class for tests of PropertyCmd classes"""

2289

1638

def runTest(self):

2290

1639

if not hasattr(self, "command"):

2291

return # Abstract TestCase class

2292

2293

if hasattr(self, "values_to_set"):

2294

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

2295

values_to_get = getattr(self, "values_to_get",

2296

self.values_to_set)

2297

else:

2298

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

2299

values_to_get = self.values_to_get

2300

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

2301

for clientpath in self.bus.clients:

2302

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

2303

Unique())

2304

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

2305

for clientpath in self.bus.clients:

2306

value = (self.bus.clients[clientpath]

2307

[self.propname])

2308

self.assertNotIsInstance(value, Unique)

2309

self.assertEqual(value_to_get, value)

2310

2311

2312

class TestEnableCmd(TestPropertySetterCmd):

2313

command = command.Enable

2314

propname = "Enabled"

2315

values_to_get = [True]

2316

2317

2318

class TestDisableCmd(TestPropertySetterCmd):

2319

command = command.Disable

2320

propname = "Enabled"

2321

values_to_get = [False]

2322

2323

2324

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2325

command = command.BumpTimeout

1640

return

1641

values_to_get = getattr(self, "values_to_get",

1642

self.values_to_set)

1643

for value_to_set, value_to_get in zip(self.values_to_set,

1644

values_to_get):

1645

for clientpath in self.clients:

1646

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

1647

old_value = client.attributes[self.propname]

1648

self.assertNotIsInstance(old_value, self.Unique)

1649

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

1650

self.run_command(value_to_set, self.clients)

1651

for clientpath in self.clients:

1652

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

1653

value = client.attributes[self.propname]

1654

self.assertNotIsInstance(value, self.Unique)

1655

self.assertEqual(value, value_to_get)

1656

1657

class Unique(object):

1658

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

1659

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

1660

1661

def run_command(self, value, clients):

1662

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

1663

1664

1665

class TestEnableCmd(TestPropertyCmd):

1666

command = EnableCmd

1667

propname = "Enabled"

1668

values_to_set = [dbus.Boolean(True)]

1669

1670

1671

class TestDisableCmd(TestPropertyCmd):

1672

command = DisableCmd

1673

propname = "Enabled"

1674

values_to_set = [dbus.Boolean(False)]

1675

1676

1677

class TestBumpTimeoutCmd(TestPropertyCmd):

1678

command = BumpTimeoutCmd

2326

1679

propname = "LastCheckedOK"

2327

values_to_get = [""]

2328

2329

2330

class TestStartCheckerCmd(TestPropertySetterCmd):

2331

command = command.StartChecker

2332

propname = "CheckerRunning"

2333

values_to_get = [True]

2334

2335

2336

class TestStopCheckerCmd(TestPropertySetterCmd):

2337

command = command.StopChecker

2338

propname = "CheckerRunning"

2339

values_to_get = [False]

2340

2341

2342

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2343

command = command.ApproveByDefault

2344

propname = "ApprovedByDefault"

2345

values_to_get = [True]

2346

2347

2348

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2349

command = command.DenyByDefault

2350

propname = "ApprovedByDefault"

2351

values_to_get = [False]

2352

2353

2354

class TestSetCheckerCmd(TestPropertySetterCmd):

2355

command = command.SetChecker

1680

values_to_set = [""]

1681

1682

1683

class TestStartCheckerCmd(TestPropertyCmd):

1684

command = StartCheckerCmd

1685

propname = "CheckerRunning"

1686

values_to_set = [dbus.Boolean(True)]

1687

1688

1689

class TestStopCheckerCmd(TestPropertyCmd):

1690

command = StopCheckerCmd

1691

propname = "CheckerRunning"

1692

values_to_set = [dbus.Boolean(False)]

1693

1694

1695

class TestApproveByDefaultCmd(TestPropertyCmd):

1696

command = ApproveByDefaultCmd

1697

propname = "ApprovedByDefault"

1698

values_to_set = [dbus.Boolean(True)]

1699

1700

1701

class TestDenyByDefaultCmd(TestPropertyCmd):

1702

command = DenyByDefaultCmd

1703

propname = "ApprovedByDefault"

1704

values_to_set = [dbus.Boolean(False)]

1705

1706

1707

class TestPropertyValueCmd(TestPropertyCmd):

1708

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

1709

1710

def runTest(self):

1711

if type(self) is TestPropertyValueCmd:

1712

return

1713

return super(TestPropertyValueCmd, self).runTest()

1714

1715

def run_command(self, value, clients):

1716

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

1717

1718

1719

class TestSetCheckerCmd(TestPropertyValueCmd):

1720

command = SetCheckerCmd

2356

1721

propname = "Checker"

2357

1722

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

2358

1723

2359

1724

2360

class TestSetHostCmd(TestPropertySetterCmd):

2361

command = command.SetHost

1725

class TestSetHostCmd(TestPropertyValueCmd):

1726

command = SetHostCmd

2362

1727

propname = "Host"

2363

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

2364

2365

2366

class TestSetSecretCmd(TestPropertySetterCmd):

2367

command = command.SetSecret

1728

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

1729

1730

1731

class TestSetSecretCmd(TestPropertyValueCmd):

1732

command = SetSecretCmd

2368

1733

propname = "Secret"

2369

1734

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

2370

1735

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

2371

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

2372

2373

2374

class TestSetTimeoutCmd(TestPropertySetterCmd):

2375

command = command.SetTimeout

1736

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

1737

1738

1739

class TestSetTimeoutCmd(TestPropertyValueCmd):

1740

command = SetTimeoutCmd

2376

1741

propname = "Timeout"

2377

1742

values_to_set = [datetime.timedelta(),

2378

1743

datetime.timedelta(minutes=5),

2379

1744

datetime.timedelta(seconds=1),

2380

1745

datetime.timedelta(weeks=1),

2381

1746

datetime.timedelta(weeks=52)]

2382

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

2383

2384

2385

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2386

command = command.SetExtendedTimeout

1747

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

1748

1749

1750

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1751

command = SetExtendedTimeoutCmd

2387

1752

propname = "ExtendedTimeout"

2388

1753

values_to_set = [datetime.timedelta(),

2389

1754

datetime.timedelta(minutes=5),

2390

1755

datetime.timedelta(seconds=1),

2391

1756

datetime.timedelta(weeks=1),

2392

1757

datetime.timedelta(weeks=52)]

2393

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

2394

2395

2396

class TestSetIntervalCmd(TestPropertySetterCmd):

2397

command = command.SetInterval

1758

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

1759

1760

1761

class TestSetIntervalCmd(TestPropertyValueCmd):

1762

command = SetIntervalCmd

2398

1763

propname = "Interval"

2399

1764

values_to_set = [datetime.timedelta(),

2400

1765

datetime.timedelta(minutes=5),

2401

1766

datetime.timedelta(seconds=1),

2402

1767

datetime.timedelta(weeks=1),

2403

1768

datetime.timedelta(weeks=52)]

2404

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

2405

2406

2407

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2408

command = command.SetApprovalDelay

1769

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

1770

1771

1772

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1773

command = SetApprovalDelayCmd

2409

1774

propname = "ApprovalDelay"

2410

1775

values_to_set = [datetime.timedelta(),

2411

1776

datetime.timedelta(minutes=5),

2412

1777

datetime.timedelta(seconds=1),

2413

1778

datetime.timedelta(weeks=1),

2414

1779

datetime.timedelta(weeks=52)]

2415

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

2416

2417

2418

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2419

command = command.SetApprovalDuration

1780

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

1781

1782

1783

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1784

command = SetApprovalDurationCmd

2420

1785

propname = "ApprovalDuration"

2421

1786

values_to_set = [datetime.timedelta(),

2422

1787

datetime.timedelta(minutes=5),

2423

1788

datetime.timedelta(seconds=1),

2424

1789

datetime.timedelta(weeks=1),

2425

1790

datetime.timedelta(weeks=52)]

2426

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

1791

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

2427

1792

2428

1793

2429

1794

Older »