/mandos/release : revision 240

1

#!/usr/bin/python3 -bbI

2

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

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

# Mandos is free software: you can redistribute it and/or modify it

12

# under the terms of the GNU General Public License as published by

13

# the Free Software Foundation, either version 3 of the License, or

14

# (at your option) any later version.

15

#

16

# Mandos is distributed in the hope that it will be useful, but

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19

# GNU General Public License for more details.

20

#

21

# You should have received a copy of the GNU General Public License

22

# along with Mandos. If not, see <http://www.gnu.org/licenses/>.

23

#

24

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

25

#

26

from __future__ import (division, absolute_import, print_function,

27

unicode_literals)

28

29

try:

30

from future_builtins import *

31

except ImportError:

32

pass

33

34

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

39

import locale

40

import datetime

41

import re

42

import collections

43

import json

44

import io

45

import tempfile

46

import contextlib

47

48

if sys.version_info.major == 2:

49

__metaclass__ = type

50

str = unicode

51

input = raw_input

52

53

54

class gi:

55

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

56

class repository:

57

class GLib:

58

class Error(Exception):

59

pass

60

61

62

dbussy = None

63

ravel = None

64

dbus_python = None

65

pydbus = None

66

67

try:

68

import dbussy

69

import ravel

70

except ImportError:

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

77

78

# Show warnings by default

79

if not sys.warnoptions:

80

import warnings

81

warnings.simplefilter("default")

82

83

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

84

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

85

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

86

87

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

88

89

if sys.version_info.major == 2:

90

import StringIO

91

io.StringIO = StringIO.StringIO

92

93

locale.setlocale(locale.LC_ALL, "")

94

95

version = "1.8.18"

96

97

98

def main():

99

parser = argparse.ArgumentParser()

100

add_command_line_options(parser)

101

102

options = parser.parse_args()

103

check_option_syntax(parser, options)

104

105

clientnames = options.client

106

107

if options.debug:

108

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

109

110

if dbussy is not None and ravel is not None:

111

bus = dbussy_adapter.CachingBus(dbussy, ravel)

112

elif pydbus is not None:

113

bus = pydbus_adapter.CachingBus(pydbus)

114

else:

115

bus = dbus_python_adapter.CachingBus(dbus_python)

116

117

try:

118

all_clients = bus.get_clients_and_properties()

119

except dbus.ConnectFailed as e:

120

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

121

sys.exit(1)

122

except dbus.Error as e:

123

log.critical(

124

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

125

sys.exit(1)

126

127

# Compile dict of (clientpath: properties) to process

128

if not clientnames:

129

clients = all_clients

130

else:

131

clients = {}

132

for name in clientnames:

133

for objpath, properties in all_clients.items():

134

if properties["Name"] == name:

135

clients[objpath] = properties

136

break

137

else:

138

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

139

sys.exit(1)

140

141

commands = commands_from_options(options)

142

143

for command in commands:

144

command.run(clients, bus)

145

146

147

def add_command_line_options(parser):

148

parser.add_argument("--version", action="version",

149

version="%(prog)s {}".format(version),

150

help="show version number and exit")

151

parser.add_argument("-a", "--all", action="store_true",

152

help="Select all clients")

153

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

154

help="Print all fields")

155

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

156

action="append_const", default=[],

157

const=command.DumpJSON(),

158

help="Dump client data in JSON format")

159

enable_disable = parser.add_mutually_exclusive_group()

160

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

161

action="append_const", default=[],

162

const=command.Enable(),

163

help="Enable client")

164

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

165

action="append_const", default=[],

166

const=command.Disable(),

167

help="disable client")

168

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

169

action="append_const", default=[],

170

const=command.BumpTimeout(),

171

help="Bump timeout for client")

172

start_stop_checker = parser.add_mutually_exclusive_group()

173

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

174

dest="commands",

175

action="append_const", default=[],

176

const=command.StartChecker(),

177

help="Start checker for client")

178

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

179

action="append_const", default=[],

180

const=command.StopChecker(),

181

help="Stop checker for client")

182

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

183

action="append_const", default=[],

184

const=command.IsEnabled(),

185

help="Check if client is enabled")

186

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

187

action="append_const", default=[],

188

const=command.Remove(),

189

help="Remove client")

190

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

191

action="append", default=[],

192

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

193

help="Set checker command for client")

194

parser.add_argument(

195

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

196

default=[], metavar="TIME",

197

type=command.SetTimeout.argparse(string_to_delta),

198

help="Set timeout for client")

199

parser.add_argument(

200

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

201

default=[], metavar="TIME",

202

type=command.SetExtendedTimeout.argparse(string_to_delta),

203

help="Set extended timeout for client")

204

parser.add_argument(

205

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

206

default=[], metavar="TIME",

207

type=command.SetInterval.argparse(string_to_delta),

208

help="Set checker interval for client")

209

approve_deny_default = parser.add_mutually_exclusive_group()

210

approve_deny_default.add_argument(

211

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

212

action="append_const", default=[],

213

const=command.ApproveByDefault(),

214

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

215

approve_deny_default.add_argument(

216

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

217

action="append_const", default=[],

218

const=command.DenyByDefault(),

219

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

220

parser.add_argument(

221

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

222

default=[], metavar="TIME",

223

type=command.SetApprovalDelay.argparse(string_to_delta),

224

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

225

parser.add_argument(

226

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

227

default=[], metavar="TIME",

228

type=command.SetApprovalDuration.argparse(string_to_delta),

229

help="Set duration of one client approval")

230

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

231

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

232

type=command.SetHost,

233

help="Set host for client")

234

parser.add_argument(

235

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

236

default=[], metavar="FILENAME",

237

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

238

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

239

approve_deny = parser.add_mutually_exclusive_group()

240

approve_deny.add_argument(

241

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

242

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

243

help="Approve any current client request")

244

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

245

action="append_const", default=[],

246

const=command.Deny(),

247

help="Deny any current client request")

248

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

249

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

250

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

251

help="Run self-test")

252

parser.add_argument("client", nargs="*", help="Client name")

253

254

255

def string_to_delta(interval):

256

"""Parse a string and return a datetime.timedelta"""

257

258

try:

259

return rfc3339_duration_to_delta(interval)

260

except ValueError as e:

261

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

262

" ".join(e.args))

263

return parse_pre_1_6_1_interval(interval)

264

265

266

def rfc3339_duration_to_delta(duration):

267

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

268

269

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

270

True

271

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

272

True

273

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

274

True

275

>>> # 60 months

276

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

277

True

278

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

279

True

280

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

281

True

282

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

283

True

284

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

285

True

286

>>> # Can not be empty:

287

>>> rfc3339_duration_to_delta("")

288

Traceback (most recent call last):

289

...

290

ValueError: Invalid RFC 3339 duration: ""

291

>>> # Must start with "P":

292

>>> rfc3339_duration_to_delta("1D")

293

Traceback (most recent call last):

294

...

295

ValueError: Invalid RFC 3339 duration: "1D"

296

>>> # Must use correct order

297

>>> rfc3339_duration_to_delta("PT1S2M")

298

Traceback (most recent call last):

299

...

300

ValueError: Invalid RFC 3339 duration: "PT1S2M"

301

>>> # Time needs time marker

302

>>> rfc3339_duration_to_delta("P1H2S")

303

Traceback (most recent call last):

304

...

305

ValueError: Invalid RFC 3339 duration: "P1H2S"

306

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

307

>>> rfc3339_duration_to_delta("P1D2W")

308

Traceback (most recent call last):

309

...

310

ValueError: Invalid RFC 3339 duration: "P1D2W"

311

>>> rfc3339_duration_to_delta("P2W2H")

312

Traceback (most recent call last):

313

...

314

ValueError: Invalid RFC 3339 duration: "P2W2H"

315

"""

316

317

# Parsing an RFC 3339 duration with regular expressions is not

318

# possible - there would have to be multiple places for the same

319

# values, like seconds. The current code, while more esoteric, is

320

# cleaner without depending on a parsing library. If Python had a

321

# built-in library for parsing we would use it, but we'd like to

322

# avoid excessive use of external libraries.

323

324

# New type for defining tokens, syntax, and semantics all-in-one

325

Token = collections.namedtuple("Token", (

326

"regexp", # To match token; if "value" is not None, must have

327

# a "group" containing digits

328

"value", # datetime.timedelta or None

329

"followers")) # Tokens valid after this token

330

# RFC 3339 "duration" tokens, syntax, and semantics; taken from

331

# the "duration" ABNF definition in RFC 3339, Appendix A.

332

token_end = Token(re.compile(r"$"), None, frozenset())

333

token_second = Token(re.compile(r"(\d+)S"),

334

datetime.timedelta(seconds=1),

335

frozenset((token_end, )))

336

token_minute = Token(re.compile(r"(\d+)M"),

337

datetime.timedelta(minutes=1),

338

frozenset((token_second, token_end)))

339

token_hour = Token(re.compile(r"(\d+)H"),

340

datetime.timedelta(hours=1),

341

frozenset((token_minute, token_end)))

342

token_time = Token(re.compile(r"T"),

343

None,

344

frozenset((token_hour, token_minute,

345

token_second)))

346

token_day = Token(re.compile(r"(\d+)D"),

347

datetime.timedelta(days=1),

348

frozenset((token_time, token_end)))

349

token_month = Token(re.compile(r"(\d+)M"),

350

datetime.timedelta(weeks=4),

351

frozenset((token_day, token_end)))

352

token_year = Token(re.compile(r"(\d+)Y"),

353

datetime.timedelta(weeks=52),

354

frozenset((token_month, token_end)))

355

token_week = Token(re.compile(r"(\d+)W"),

356

datetime.timedelta(weeks=1),

357

frozenset((token_end, )))

358

token_duration = Token(re.compile(r"P"), None,

359

frozenset((token_year, token_month,

360

token_day, token_time,

361

token_week)))

362

# Define starting values:

363

# Value so far

364

value = datetime.timedelta()

365

found_token = None

366

# Following valid tokens

367

followers = frozenset((token_duration, ))

368

# String left to parse

369

s = duration

370

# Loop until end token is found

371

while found_token is not token_end:

372

# Search for any currently valid tokens

373

for token in followers:

374

match = token.regexp.match(s)

375

if match is not None:

376

# Token found

377

if token.value is not None:

378

# Value found, parse digits

379

factor = int(match.group(1), 10)

380

# Add to value so far

381

value += factor * token.value

382

# Strip token from string

383

s = token.regexp.sub("", s, 1)

384

# Go to found token

385

found_token = token

386

# Set valid next tokens

387

followers = found_token.followers

388

break

389

else:

390

# No currently valid tokens were found

391

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

392

.format(duration))

393

# End token found

394

return value

395

396

397

def parse_pre_1_6_1_interval(interval):

398

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

399

and return a datetime.timedelta

400

401

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

402

True

403

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

404

True

405

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

406

True

407

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

408

True

409

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

410

True

411

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

412

True

413

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

414

True

415

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

416

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \

417

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

418

True

419

420

"""

421

422

value = datetime.timedelta(0)

423

regexp = re.compile(r"(\d+)([dsmhw]?)")

424

425

for num, suffix in regexp.findall(interval):

426

if suffix == "d":

427

value += datetime.timedelta(int(num))

428

elif suffix == "s":

429

value += datetime.timedelta(0, int(num))

430

elif suffix == "m":

431

value += datetime.timedelta(0, 0, 0, 0, int(num))

432

elif suffix == "h":

433

value += datetime.timedelta(0, 0, 0, 0, 0, int(num))

434

elif suffix == "w":

435

value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))

436

elif suffix == "":

437

value += datetime.timedelta(0, 0, 0, int(num))

438

return value

439

440

441

def check_option_syntax(parser, options):

442

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

443

argparse"""

444

445

def has_commands(options, commands=None):

446

if commands is None:

447

commands = (command.Enable,

448

command.Disable,

449

command.BumpTimeout,

450

command.StartChecker,

451

command.StopChecker,

452

command.IsEnabled,

453

command.Remove,

454

command.SetChecker,

455

command.SetTimeout,

456

command.SetExtendedTimeout,

457

command.SetInterval,

458

command.ApproveByDefault,

459

command.DenyByDefault,

460

command.SetApprovalDelay,

461

command.SetApprovalDuration,

462

command.SetHost,

463

command.SetSecret,

464

command.Approve,

465

command.Deny)

466

return any(isinstance(cmd, commands)

467

for cmd in options.commands)

468

469

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

470

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

471

if options.verbose and has_commands(options):

472

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

473

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

474

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

475

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

476

if options.all and not has_commands(options):

477

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

478

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

479

and len(options.client) > 1):

480

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

481

if (len(options.commands) > 1

482

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

483

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

484

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

485

486

487

class dbus:

488

489

class SystemBus:

490

491

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

493

def get_managed_objects(self, busname, objectpath):

494

return self.call_method("GetManagedObjects", busname,

495

objectpath,

496

self.object_manager_iface)

497

498

properties_iface = "org.freedesktop.DBus.Properties"

499

500

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

501

value):

502

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

503

self.properties_iface, interface, key,

504

value)

505

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

509

510

class MandosBus(SystemBus):

511

busname_domain = "se.recompile"

512

busname = busname_domain + ".Mandos"

513

server_path = "/"

514

server_interface = busname_domain + ".Mandos"

515

client_interface = busname_domain + ".Mandos.Client"

516

del busname_domain

517

518

def get_clients_and_properties(self):

519

managed_objects = self.get_managed_objects(

520

self.busname, self.server_path)

521

return {objpath: properties[self.client_interface]

522

for objpath, properties in managed_objects.items()

523

if self.client_interface in properties}

524

525

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

526

return self.set_property(self.busname, objectpath,

527

self.client_interface, key,

528

value)

529

530

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

531

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

532

self.client_interface, *args)

533

534

def call_server_method(self, method, *args):

535

return self.call_method(method, self.busname,

536

self.server_path,

537

self.server_interface, *args)

538

539

class Error(Exception):

540

pass

541

542

class ConnectFailed(Error):

543

pass

544

545

546

class dbus_python_adapter:

547

548

class SystemBus(dbus.MandosBus):

549

"""Use dbus-python"""

550

551

def __init__(self, module=dbus_python):

552

self.dbus_python = module

553

self.bus = self.dbus_python.SystemBus()

554

555

@contextlib.contextmanager

556

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

557

try:

558

yield

559

except self.dbus_python.exceptions.DBusException as e:

560

# This does what "raise from" would do

561

exc = exception_class(*e.args)

562

exc.__cause__ = e

563

raise exc

564

565

def call_method(self, methodname, busname, objectpath,

566

interface, *args):

567

proxy_object = self.get_object(busname, objectpath)

568

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

569

interface, methodname,

570

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

571

method = getattr(proxy_object, methodname)

572

with self.convert_exception():

573

with dbus_python_adapter.SilenceLogger(

574

"dbus.proxies"):

575

value = method(*args, dbus_interface=interface)

576

return self.type_filter(value)

577

578

def get_object(self, busname, objectpath):

579

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

580

busname, objectpath)

581

with self.convert_exception(dbus.ConnectFailed):

582

return self.bus.get_object(busname, objectpath)

583

584

def type_filter(self, value):

585

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

586

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

587

return bool(value)

588

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

589

return str(value)

590

# Also recurse into dictionaries

591

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

592

return {self.type_filter(key):

593

self.type_filter(subval)

594

for key, subval in value.items()}

595

return value

596

597

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

598

if key == "Secret":

599

if not isinstance(value, bytes):

600

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

601

value = self.dbus_python.ByteArray(value)

602

return self.set_property(self.busname, objectpath,

603

self.client_interface, key,

604

value)

605

606

class SilenceLogger:

607

"Simple context manager to silence a particular logger"

608

609

def __init__(self, loggername):

610

self.logger = logging.getLogger(loggername)

611

612

def __enter__(self):

613

self.logger.addFilter(self.nullfilter)

614

615

class NullFilter(logging.Filter):

616

def filter(self, record):

617

return False

618

619

nullfilter = NullFilter()

620

621

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

622

self.logger.removeFilter(self.nullfilter)

623

624

class CachingBus(SystemBus):

625

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

626

627

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

628

self.object_cache = {}

629

super(dbus_python_adapter.CachingBus,

630

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

631

632

def get_object(self, busname, objectpath):

633

try:

634

return self.object_cache[(busname, objectpath)]

635

except KeyError:

636

new_object = super(

637

dbus_python_adapter.CachingBus,

638

self).get_object(busname, objectpath)

639

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

640

return new_object

641

642

643

class pydbus_adapter:

644

class SystemBus(dbus.MandosBus):

645

def __init__(self, module=pydbus):

646

self.pydbus = module

647

self.bus = self.pydbus.SystemBus()

648

649

@contextlib.contextmanager

650

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

651

try:

652

yield

653

except gi.repository.GLib.Error as e:

654

# This does what "raise from" would do

655

exc = exception_class(*e.args)

656

exc.__cause__ = e

657

raise exc

658

659

def call_method(self, methodname, busname, objectpath,

660

interface, *args):

661

proxy_object = self.get(busname, objectpath)

662

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

663

interface, methodname,

664

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

665

method = getattr(proxy_object[interface], methodname)

666

with self.convert_exception():

667

return method(*args)

668

669

def get(self, busname, objectpath):

670

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

671

busname, objectpath)

672

with self.convert_exception(dbus.ConnectFailed):

673

if sys.version_info.major <= 2:

674

with warnings.catch_warnings():

675

warnings.filterwarnings(

676

"ignore", "", DeprecationWarning,

677

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

678

return self.bus.get(busname, objectpath)

679

else:

680

return self.bus.get(busname, objectpath)

681

682

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

683

value):

684

proxy_object = self.get(busname, objectpath)

685

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

686

objectpath, self.properties_iface, interface,

687

key, value)

688

setattr(proxy_object[interface], key, value)

689

690

class CachingBus(SystemBus):

691

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

692

693

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

694

self.object_cache = {}

695

super(pydbus_adapter.CachingBus,

696

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

697

698

def get(self, busname, objectpath):

699

try:

700

return self.object_cache[(busname, objectpath)]

701

except KeyError:

702

new_object = (super(pydbus_adapter.CachingBus, self)

703

.get(busname, objectpath))

704

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

705

return new_object

706

707

708

class dbussy_adapter:

709

class SystemBus(dbus.SystemBus):

710

"""Use DBussy"""

711

712

def __init__(self, dbussy, ravel):

713

self.dbussy = dbussy

714

self.ravel = ravel

715

self.bus = ravel.system_bus()

716

717

@contextlib.contextmanager

718

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

719

try:

720

yield

721

except self.dbussy.DBusError as e:

722

# This does what "raise from" would do

723

exc = exception_class(*e.args)

724

exc.__cause__ = e

725

raise exc

726

727

def call_method(self, methodname, busname, objectpath,

728

interface, *args):

729

proxy_object = self.get_object(busname, objectpath)

730

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

731

interface, methodname,

732

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

733

iface = proxy_object.get_interface(interface)

734

method = getattr(iface, methodname)

735

with self.convert_exception(dbus.Error):

736

value = method(*args)

737

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

738

# list of one element with the return value

739

if value:

740

return self.type_filter(value[0])

741

742

def get_object(self, busname, objectpath):

743

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

744

busname, objectpath)

745

with self.convert_exception(dbus.ConnectFailed):

746

return self.bus[busname][objectpath]

747

748

def type_filter(self, value):

749

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

750

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

751

# Tuple[dbussy.DBUS.Signature, Any]

752

if isinstance(value, tuple):

753

if (len(value) == 2

754

and isinstance(value[0],

755

self.dbussy.DBUS.Signature)):

756

return self.type_filter(value[1])

757

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

758

return str(value)

759

# Also recurse into dictionaries

760

elif isinstance(value, dict):

761

return {self.type_filter(key):

762

self.type_filter(subval)

763

for key, subval in value.items()}

764

return value

765

766

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

767

value):

768

proxy_object = self.get_object(busname, objectpath)

769

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

770

objectpath, self.properties_iface, interface,

771

key, value)

772

if key == "Secret":

773

# DBussy wants a Byte Array to be a sequence of

774

# values, not a byte string

775

value = tuple(value)

776

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

777

778

class MandosBus(SystemBus, dbus.MandosBus):

779

pass

780

781

class CachingBus(MandosBus):

782

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

783

784

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

785

self.object_cache = {}

786

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

787

**kwargs)

788

789

def get_object(self, busname, objectpath):

790

try:

791

return self.object_cache[(busname, objectpath)]

792

except KeyError:

793

new_object = super(

794

dbussy_adapter.CachingBus,

795

self).get_object(busname, objectpath)

796

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

797

return new_object

798

799

800

def commands_from_options(options):

801

802

commands = list(options.commands)

803

804

def find_cmd(cmd, commands):

805

i = 0

806

for i, c in enumerate(commands):

807

if isinstance(c, cmd):

808

return i

809

return i+1

810

811

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

812

# to occur ahead of command.Remove.

813

index_of_remove = find_cmd(command.Remove, commands)

814

before_remove = commands[:index_of_remove]

815

after_remove = commands[index_of_remove:]

816

cleaned_after = []

817

for cmd in after_remove:

818

if isinstance(cmd, command.Deny):

819

before_remove.append(cmd)

820

else:

821

cleaned_after.append(cmd)

822

if cleaned_after != after_remove:

823

commands = before_remove + cleaned_after

824

825

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

826

# optionally verbosely

827

if not commands:

828

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

829

830

return commands

831

832

833

class command:

834

"""A namespace for command classes"""

835

836

class Base:

837

"""Abstract base class for commands"""

838

839

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

840

"""Normal commands should implement run_on_one_client(),

841

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

842

override this run() method instead.

843

"""

844

self.bus = bus

845

for client, properties in clients.items():

846

self.run_on_one_client(client, properties)

847

848

class IsEnabled(Base):

849

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

850

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

851

if properties["Enabled"]:

852

sys.exit(0)

853

sys.exit(1)

854

855

class Approve(Base):

856

def run_on_one_client(self, client, properties):

857

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

858

859

class Deny(Base):

860

def run_on_one_client(self, client, properties):

861

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

862

863

class Remove(Base):

864

def run(self, clients, bus):

865

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

866

bus.call_server_method("RemoveClient", clientpath)

867

868

class Output(Base):

869

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

870

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

871

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

872

"Fingerprint", "CheckerRunning",

873

"LastEnabled", "ApprovalPending",

874

"ApprovedByDefault", "LastApprovalRequest",

875

"ApprovalDelay", "ApprovalDuration",

876

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

877

"LastCheckerStatus")

878

879

class DumpJSON(Output):

880

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

881

data = {properties["Name"]:

882

{key: properties[key]

883

for key in self.all_keywords}

884

for properties in clients.values()}

885

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

886

887

class PrintTable(Output):

888

def __init__(self, verbose=False):

889

self.verbose = verbose

890

891

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

892

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

893

"LastCheckedOK")

894

keywords = default_keywords

895

if self.verbose:

896

keywords = self.all_keywords

897

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

898

899

class TableOfClients:

900

tableheaders = {

901

"Name": "Name",

902

"Enabled": "Enabled",

903

"Timeout": "Timeout",

904

"LastCheckedOK": "Last Successful Check",

905

"LastApprovalRequest": "Last Approval Request",

906

"Created": "Created",

907

"Interval": "Interval",

908

"Host": "Host",

909

"Fingerprint": "Fingerprint",

910

"KeyID": "Key ID",

911

"CheckerRunning": "Check Is Running",

912

"LastEnabled": "Last Enabled",

913

"ApprovalPending": "Approval Is Pending",

914

"ApprovedByDefault": "Approved By Default",

915

"ApprovalDelay": "Approval Delay",

916

"ApprovalDuration": "Approval Duration",

917

"Checker": "Checker",

918

"ExtendedTimeout": "Extended Timeout",

919

"Expires": "Expires",

920

"LastCheckerStatus": "Last Checker Status",

921

}

922

923

def __init__(self, clients, keywords):

924

self.clients = clients

925

self.keywords = keywords

926

927

def __str__(self):

928

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

929

930

if sys.version_info.major == 2:

931

__unicode__ = __str__

932

933

def __str__(self):

934

return str(self).encode(

935

locale.getpreferredencoding())

936

937

def rows(self):

938

format_string = self.row_formatting_string()

939

rows = [self.header_line(format_string)]

940

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

941

for client in self.clients)

942

return rows

943

944

def row_formatting_string(self):

945

"Format string used to format table rows"

946

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

947

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

948

*(len(self.string_from_client(client,

949

key))

950

for client in self.clients)),

951

key=key)

952

for key in self.keywords)

953

954

def string_from_client(self, client, key):

955

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

956

957

@classmethod

958

def valuetostring(cls, value, keyword):

959

if isinstance(value, bool):

960

return "Yes" if value else "No"

961

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

962

"ApprovalDuration", "ExtendedTimeout"):

963

return cls.milliseconds_to_string(value)

964

return str(value)

965

966

def header_line(self, format_string):

967

return format_string.format(**self.tableheaders)

968

969

def client_line(self, client, format_string):

970

return format_string.format(

971

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

972

for key in self.keywords})

973

974

@staticmethod

975

def milliseconds_to_string(ms):

976

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

977

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

978

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

979

if td.days else "",

980

hours=td.seconds // 3600,

981

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

982

seconds=td.seconds % 60))

983

984

class PropertySetter(Base):

985

"Abstract class for Actions for setting one client property"

986

987

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

988

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

989

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

990

self.value_to_set)

991

992

@property

993

def propname(self):

994

raise NotImplementedError()

995

996

class Enable(PropertySetter):

997

propname = "Enabled"

998

value_to_set = True

999

1000

class Disable(PropertySetter):

1001

propname = "Enabled"

1002

value_to_set = False

1003

1004

class BumpTimeout(PropertySetter):

1005

propname = "LastCheckedOK"

1006

value_to_set = ""

1007

1008

class StartChecker(PropertySetter):

1009

propname = "CheckerRunning"

1010

value_to_set = True

1011

1012

class StopChecker(PropertySetter):

1013

propname = "CheckerRunning"

1014

value_to_set = False

1015

1016

class ApproveByDefault(PropertySetter):

1017

propname = "ApprovedByDefault"

1018

value_to_set = True

1019

1020

class DenyByDefault(PropertySetter):

1021

propname = "ApprovedByDefault"

1022

value_to_set = False

1023

1024

class PropertySetterValue(PropertySetter):

1025

"""Abstract class for PropertySetter recieving a value as

1026

constructor argument instead of a class attribute."""

1027

1028

def __init__(self, value):

1029

self.value_to_set = value

1030

1031

@classmethod

1032

def argparse(cls, argtype):

1033

def cmdtype(arg):

1034

return cls(argtype(arg))

1035

return cmdtype

1036

1037

class SetChecker(PropertySetterValue):

1038

propname = "Checker"

1039

1040

class SetHost(PropertySetterValue):

1041

propname = "Host"

1042

1043

class SetSecret(PropertySetterValue):

1044

propname = "Secret"

1045

1046

@property

1047

def value_to_set(self):

1048

return self._vts

1049

1050

@value_to_set.setter

1051

def value_to_set(self, value):

1052

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

1053

self._vts = value.read()

1054

value.close()

1055

1056

class PropertySetterValueMilliseconds(PropertySetterValue):

1057

"""Abstract class for PropertySetterValue taking a value

1058

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

1059

milliseconds."""

1060

1061

@property

1062

def value_to_set(self):

1063

return self._vts

1064

1065

@value_to_set.setter

1066

def value_to_set(self, value):

1067

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

1068

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

1069

1070

class SetTimeout(PropertySetterValueMilliseconds):

1071

propname = "Timeout"

1072

1073

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1074

propname = "ExtendedTimeout"

1075

1076

class SetInterval(PropertySetterValueMilliseconds):

1077

propname = "Interval"

1078

1079

class SetApprovalDelay(PropertySetterValueMilliseconds):

1080

propname = "ApprovalDelay"

1081

1082

class SetApprovalDuration(PropertySetterValueMilliseconds):

1083

propname = "ApprovalDuration"

1084

1085

1086

1087

class TestCaseWithAssertLogs(unittest.TestCase):

1088

"""unittest.TestCase.assertLogs only exists in Python 3.4"""

1089

1090

if not hasattr(unittest.TestCase, "assertLogs"):

1091

@contextlib.contextmanager

1092

def assertLogs(self, logger, level=logging.INFO):

1093

capturing_handler = self.CapturingLevelHandler(level)

1094

old_level = logger.level

1095

old_propagate = logger.propagate

1096

logger.addHandler(capturing_handler)

1097

logger.setLevel(level)

1098

logger.propagate = False

1099

try:

1100

yield capturing_handler.watcher

1101

finally:

1102

logger.propagate = old_propagate

1103

logger.removeHandler(capturing_handler)

1104

logger.setLevel(old_level)

1105

self.assertGreater(len(capturing_handler.watcher.records),

1106

0)

1107

1108

class CapturingLevelHandler(logging.Handler):

1109

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

1110

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

1111

self.watcher = self.LoggingWatcher([], [])

1112

def emit(self, record):

1113

self.watcher.records.append(record)

1114

self.watcher.output.append(self.format(record))

1115

1116

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1117

("records",

1118

"output"))

1119

1120

1121

class Unique:

1122

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

1123

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

1124

1125

1126

class Test_string_to_delta(TestCaseWithAssertLogs):

1127

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

1128

# rfc3339_duration_to_delta() already has more comprehensive

1129

# tests, which are run by doctest.

1130

1131

def test_rfc3339_zero_seconds(self):

1132

self.assertEqual(datetime.timedelta(),

1133

string_to_delta("PT0S"))

1134

1135

def test_rfc3339_zero_days(self):

1136

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

1137

1138

def test_rfc3339_one_second(self):

1139

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

1140

string_to_delta("PT1S"))

1141

1142

def test_rfc3339_two_hours(self):

1143

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

1144

string_to_delta("PT2H"))

1145

1146

def test_falls_back_to_pre_1_6_1_with_warning(self):

1147

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

1148

value = string_to_delta("2h")

1149

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

1150

1151

1152

class Test_check_option_syntax(unittest.TestCase):

1153

def setUp(self):

1154

self.parser = argparse.ArgumentParser()

1155

add_command_line_options(self.parser)

1156

1157

def test_actions_requires_client_or_all(self):

1158

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

1159

args = self.actionargs(action, value)

1160

with self.assertParseError():

1161

self.parse_args(args)

1162

1163

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

1164

# check_option_syntax()

1165

actions = {

1166

"--enable": None,

1167

"--disable": None,

1168

"--bump-timeout": None,

1169

"--start-checker": None,

1170

"--stop-checker": None,

1171

"--is-enabled": None,

1172

"--remove": None,

1173

"--checker": "x",

1174

"--timeout": "PT0S",

1175

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

1176

"--interval": "PT0S",

1177

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

1178

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

1179

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

1180

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

1181

"--host": "hostname",

1182

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

1183

"--approve": None,

1184

"--deny": None,

1185

}

1186

1187

@staticmethod

1188

def actionargs(action, value, *args):

1189

if value is not None:

1190

return [action, value] + list(args)

1191

else:

1192

return [action] + list(args)

1193

1194

@contextlib.contextmanager

1195

def assertParseError(self):

1196

with self.assertRaises(SystemExit) as e:

1197

with self.redirect_stderr_to_devnull():

1198

yield

1199

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

1200

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

1201

# /argparse.html#exiting-methods

1202

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

1203

1204

def parse_args(self, args):

1205

options = self.parser.parse_args(args)

1206

check_option_syntax(self.parser, options)

1207

1208

@staticmethod

1209

@contextlib.contextmanager

1210

def redirect_stderr_to_devnull():

1211

old_stderr = sys.stderr

1212

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

1213

sys.stderr = null

1214

try:

1215

yield

1216

finally:

1217

sys.stderr = old_stderr

1218

1219

def check_option_syntax(self, options):

1220

check_option_syntax(self.parser, options)

1221

1222

def test_actions_all_conflicts_with_verbose(self):

1223

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

1224

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

1225

"--verbose")

1226

with self.assertParseError():

1227

self.parse_args(args)

1228

1229

def test_actions_with_client_conflicts_with_verbose(self):

1230

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

1231

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

1232

"client")

1233

with self.assertParseError():

1234

self.parse_args(args)

1235

1236

def test_dump_json_conflicts_with_verbose(self):

1237

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

1238

with self.assertParseError():

1239

self.parse_args(args)

1240

1241

def test_dump_json_conflicts_with_action(self):

1242

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

1243

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

1244

with self.assertParseError():

1245

self.parse_args(args)

1246

1247

def test_all_can_not_be_alone(self):

1248

args = ["--all"]

1249

with self.assertParseError():

1250

self.parse_args(args)

1251

1252

def test_all_is_ok_with_any_action(self):

1253

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

1254

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

1255

self.parse_args(args)

1256

1257

def test_any_action_is_ok_with_one_client(self):

1258

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

1259

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

1260

self.parse_args(args)

1261

1262

def test_one_client_with_all_actions_except_is_enabled(self):

1263

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

1264

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

1265

continue

1266

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

1267

self.parse_args(args)

1268

1269

def test_two_clients_with_all_actions_except_is_enabled(self):

1270

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

1271

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

1272

continue

1273

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

1274

"client2")

1275

self.parse_args(args)

1276

1277

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1278

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

1279

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

1280

continue

1281

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

1282

"client2")

1283

self.parse_args(args)

1284

1285

def test_is_enabled_fails_without_client(self):

1286

args = ["--is-enabled"]

1287

with self.assertParseError():

1288

self.parse_args(args)

1289

1290

def test_is_enabled_fails_with_two_clients(self):

1291

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

1292

with self.assertParseError():

1293

self.parse_args(args)

1294

1295

def test_remove_can_only_be_combined_with_action_deny(self):

1296

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

1297

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

1298

continue

1299

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

1300

"--remove")

1301

with self.assertParseError():

1302

self.parse_args(args)

1303

1304

1305

class Test_dbus_exceptions(unittest.TestCase):

1306

1307

def test_dbus_ConnectFailed_is_Error(self):

1308

with self.assertRaises(dbus.Error):

1309

raise dbus.ConnectFailed()

1310

1311

1312

class Test_dbus_MandosBus(unittest.TestCase):

1313

1314

class MockMandosBus(dbus.MandosBus):

1315

def __init__(self):

1316

self._name = "se.recompile.Mandos"

1317

self._server_path = "/"

1318

self._server_interface = "se.recompile.Mandos"

1319

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

1320

self.calls = []

1321

self.call_method_return = Unique()

1322

1323

def call_method(self, methodname, busname, objectpath,

1324

interface, *args):

1325

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

1326

interface, args))

1327

return self.call_method_return

1328

1329

def setUp(self):

1330

self.bus = self.MockMandosBus()

1331

1332

def test_set_client_property(self):

1333

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

1334

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

1335

"org.freedesktop.DBus.Properties",

1336

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

1337

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

1338

1339

def test_call_client_method(self):

1340

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

1341

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

1342

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

1343

self.bus._client_interface, ())

1344

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

1345

1346

def test_call_client_method_with_args(self):

1347

args = (Unique(), Unique())

1348

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

1349

*args)

1350

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

1351

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

1352

self.bus._client_interface,

1353

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

1354

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

1355

1356

def test_get_clients_and_properties(self):

1357

managed_objects = {

1358

"objectpath": {

1359

self.bus._client_interface: {

1360

"key": "value",

1361

"bool": True,

1362

},

1363

"irrelevant_interface": {

1364

"key": "othervalue",

1365

"bool": False,

1366

},

1367

},

1368

"other_objectpath": {

1369

"other_irrelevant_interface": {

1370

"key": "value 3",

1371

"bool": None,

1372

},

1373

},

1374

}

1375

expected_clients_and_properties = {

1376

"objectpath": {

1377

"key": "value",

1378

"bool": True,

1379

}

1380

}

1381

self.bus.call_method_return = managed_objects

1382

ret = self.bus.get_clients_and_properties()

1383

self.assertDictEqual(expected_clients_and_properties, ret)

1384

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

1385

self.bus._server_path,

1386

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

1387

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

1388

1389

def test_call_server_method(self):

1390

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

1391

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

1392

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

1393

self.bus._server_path,

1394

self.bus._server_interface, ())

1395

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

1396

1397

def test_call_server_method_with_args(self):

1398

args = (Unique(), Unique())

1399

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

1400

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

1401

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

1402

self.bus._server_path,

1403

self.bus._server_interface,

1404

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

1405

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

1406

1407

1408

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1409

1410

def MockDBusPython_func(self, func):

1411

class mock_dbus_python:

1412

"""mock dbus-python module"""

1413

class exceptions:

1414

"""Pseudo-namespace"""

1415

class DBusException(Exception):

1416

pass

1417

class SystemBus:

1418

@staticmethod

1419

def get_object(busname, objectpath):

1420

DBusObject = collections.namedtuple(

1421

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

1422

def method(*args, **kwargs):

1423

self.assertEqual({"dbus_interface":

1424

"interface"},

1425

kwargs)

1426

return func(*args)

1427

def set_property(interface, key, value,

1428

dbus_interface=None):

1429

self.assertEqual(

1430

"org.freedesktop.DBus.Properties",

1431

dbus_interface)

1432

self.assertEqual("Secret", key)

1433

return func(interface, key, value,

1434

dbus_interface=dbus_interface)

1435

return DBusObject(methodname=method,

1436

Set=set_property)

1437

class Boolean:

1438

def __init__(self, value):

1439

self.value = bool(value)

1440

def __bool__(self):

1441

return self.value

1442

if sys.version_info.major == 2:

1443

__nonzero__ = __bool__

1444

class ObjectPath(str):

1445

pass

1446

class Dictionary(dict):

1447

pass

1448

class ByteArray(bytes):

1449

pass

1450

return mock_dbus_python

1451

1452

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

1453

interface, *args):

1454

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

1455

return bus.call_method(methodname, busname, objectpath,

1456

interface, *args)

1457

1458

def test_call_method_returns(self):

1459

expected_method_return = Unique()

1460

method_args = (Unique(), Unique())

1461

def func(*args):

1462

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

1463

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

1464

self.assertIs(marg, arg)

1465

return expected_method_return

1466

mock_dbus_python = self.MockDBusPython_func(func)

1467

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1468

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

1469

"objectpath", "interface",

1470

*method_args)

1471

self.assertIs(ret, expected_method_return)

1472

1473

def test_call_method_filters_bool_true(self):

1474

def func():

1475

return method_return

1476

mock_dbus_python = self.MockDBusPython_func(func)

1477

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1478

method_return = mock_dbus_python.Boolean(True)

1479

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

1480

"objectpath", "interface")

1481

self.assertTrue(ret)

1482

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1483

1484

def test_call_method_filters_bool_false(self):

1485

def func():

1486

return method_return

1487

mock_dbus_python = self.MockDBusPython_func(func)

1488

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1489

method_return = mock_dbus_python.Boolean(False)

1490

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

1491

"objectpath", "interface")

1492

self.assertFalse(ret)

1493

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1494

1495

def test_call_method_filters_objectpath(self):

1496

def func():

1497

return method_return

1498

mock_dbus_python = self.MockDBusPython_func(func)

1499

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1500

method_return = mock_dbus_python.ObjectPath("objectpath")

1501

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

1502

"objectpath", "interface")

1503

self.assertEqual("objectpath", ret)

1504

self.assertIsNot("objectpath", ret)

1505

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1506

1507

def test_call_method_filters_booleans_in_dict(self):

1508

def func():

1509

return method_return

1510

mock_dbus_python = self.MockDBusPython_func(func)

1511

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1512

method_return = mock_dbus_python.Dictionary(

1513

{mock_dbus_python.Boolean(True):

1514

mock_dbus_python.Boolean(False),

1515

mock_dbus_python.Boolean(False):

1516

mock_dbus_python.Boolean(True)})

1517

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

1518

"objectpath", "interface")

1519

expected_method_return = {True: False,

1520

False: True}

1521

self.assertEqual(expected_method_return, ret)

1522

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1523

1524

def test_call_method_filters_objectpaths_in_dict(self):

1525

def func():

1526

return method_return

1527

mock_dbus_python = self.MockDBusPython_func(func)

1528

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1529

method_return = mock_dbus_python.Dictionary(

1530

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1531

mock_dbus_python.ObjectPath("objectpath_value_1"),

1532

mock_dbus_python.ObjectPath("objectpath_key_2"):

1533

mock_dbus_python.ObjectPath("objectpath_value_2")})

1534

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

1535

"objectpath", "interface")

1536

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

1537

for key, value in

1538

method_return.items()}

1539

self.assertEqual(expected_method_return, ret)

1540

self.assertIsInstance(ret, dict)

1541

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1542

1543

def test_call_method_filters_dict_in_dict(self):

1544

def func():

1545

return method_return

1546

mock_dbus_python = self.MockDBusPython_func(func)

1547

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1548

method_return = mock_dbus_python.Dictionary(

1549

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

1550

"key12": "value12"}),

1551

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

1552

"key22": "value22"})})

1553

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

1554

"objectpath", "interface")

1555

expected_method_return = {

1556

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

1557

"key12": "value12"},

1558

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

1559

"key22": "value22"},

1560

}

1561

self.assertEqual(expected_method_return, ret)

1562

self.assertIsInstance(ret, dict)

1563

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1564

for key, value in ret.items():

1565

self.assertIsInstance(value, dict)

1566

self.assertEqual(expected_method_return[key], value)

1567

self.assertNotIsInstance(value,

1568

mock_dbus_python.Dictionary)

1569

1570

def test_call_method_filters_dict_three_deep(self):

1571

def func():

1572

return method_return

1573

mock_dbus_python = self.MockDBusPython_func(func)

1574

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1575

method_return = mock_dbus_python.Dictionary(

1576

{"key1":

1577

mock_dbus_python.Dictionary(

1578

{"key2":

1579

mock_dbus_python.Dictionary(

1580

{"key3":

1581

mock_dbus_python.Boolean(True),

1582

}),

1583

}),

1584

})

1585

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

1586

"objectpath", "interface")

1587

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

1588

self.assertEqual(expected_method_return, ret)

1589

self.assertIsInstance(ret, dict)

1590

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1591

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

1592

self.assertNotIsInstance(ret["key1"],

1593

mock_dbus_python.Dictionary)

1594

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

1595

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

1596

mock_dbus_python.Dictionary)

1597

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

1598

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

1599

mock_dbus_python.Boolean)

1600

1601

def test_call_method_handles_exception(self):

1602

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

1603

1604

def func():

1605

dbus_logger.error("Test")

1606

raise mock_dbus_python.exceptions.DBusException()

1607

1608

mock_dbus_python = self.MockDBusPython_func(func)

1609

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1610

1611

class CountingHandler(logging.Handler):

1612

count = 0

1613

def emit(self, record):

1614

self.count += 1

1615

1616

counting_handler = CountingHandler()

1617

1618

dbus_logger.addHandler(counting_handler)

1619

1620

try:

1621

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

1622

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

1623

"objectpath", "interface")

1624

finally:

1625

dbus_logger.removeFilter(counting_handler)

1626

1627

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

1628

1629

# Make sure the dbus logger was suppressed

1630

self.assertEqual(0, counting_handler.count)

1631

1632

def test_Set_Secret_sends_bytearray(self):

1633

ret = [None]

1634

def func(*args, **kwargs):

1635

ret[0] = (args, kwargs)

1636

mock_dbus_python = self.MockDBusPython_func(func)

1637

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1638

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

1639

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

1640

mock_dbus_python.ByteArray(b"value")),

1641

{"dbus_interface":

1642

"org.freedesktop.DBus.Properties"})

1643

self.assertEqual(expected_call, ret[0])

1644

if sys.version_info.major == 2:

1645

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

1646

mock_dbus_python.ByteArray)

1647

1648

def test_get_object_converts_to_correct_exception(self):

1649

bus = dbus_python_adapter.SystemBus(

1650

self.fake_dbus_python_raises_exception_on_connect)

1651

with self.assertRaises(dbus.ConnectFailed):

1652

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

1653

"objectpath", "interface")

1654

1655

class fake_dbus_python_raises_exception_on_connect:

1656

"""fake dbus-python module"""

1657

class exceptions:

1658

"""Pseudo-namespace"""

1659

class DBusException(Exception):

1660

pass

1661

1662

@classmethod

1663

def SystemBus(cls):

1664

def get_object(busname, objectpath):

1665

raise cls.exceptions.DBusException()

1666

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

1667

return Bus(get_object=get_object)

1668

1669

1670

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1671

class mock_dbus_python:

1672

"""mock dbus-python modules"""

1673

class SystemBus:

1674

@staticmethod

1675

def get_object(busname, objectpath):

1676

return Unique()

1677

1678

def setUp(self):

1679

self.bus = dbus_python_adapter.CachingBus(

1680

self.mock_dbus_python)

1681

1682

def test_returns_distinct_objectpaths(self):

1683

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

1684

self.assertIsInstance(obj1, Unique)

1685

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

1686

self.assertIsInstance(obj2, Unique)

1687

self.assertIsNot(obj1, obj2)

1688

1689

def test_returns_distinct_busnames(self):

1690

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

1691

self.assertIsInstance(obj1, Unique)

1692

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

1693

self.assertIsInstance(obj2, Unique)

1694

self.assertIsNot(obj1, obj2)

1695

1696

def test_returns_distinct_both(self):

1697

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

1698

self.assertIsInstance(obj1, Unique)

1699

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

1700

self.assertIsInstance(obj2, Unique)

1701

self.assertIsNot(obj1, obj2)

1702

1703

def test_returns_same(self):

1704

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

1705

self.assertIsInstance(obj1, Unique)

1706

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

1707

self.assertIsInstance(obj2, Unique)

1708

self.assertIs(obj1, obj2)

1709

1710

def test_returns_same_old(self):

1711

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

1712

self.assertIsInstance(obj1, Unique)

1713

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

1714

self.assertIsInstance(obj2, Unique)

1715

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

1716

self.assertIsInstance(obj1b, Unique)

1717

self.assertIsNot(obj1, obj2)

1718

self.assertIsNot(obj2, obj1b)

1719

self.assertIs(obj1, obj1b)

1720

1721

1722

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1723

1724

def Stub_pydbus_func(self, func):

1725

class stub_pydbus:

1726

"""stub pydbus module"""

1727

class SystemBus:

1728

@staticmethod

1729

def get(busname, objectpath):

1730

DBusObject = collections.namedtuple(

1731

"DBusObject", ("methodname",))

1732

return {"interface":

1733

DBusObject(methodname=func)}

1734

return stub_pydbus

1735

1736

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

1737

interface, *args):

1738

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

1739

return bus.call_method(methodname, busname, objectpath,

1740

interface, *args)

1741

1742

def test_call_method_returns(self):

1743

expected_method_return = Unique()

1744

method_args = (Unique(), Unique())

1745

def func(*args):

1746

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

1747

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

1748

self.assertIs(marg, arg)

1749

return expected_method_return

1750

stub_pydbus = self.Stub_pydbus_func(func)

1751

bus = pydbus_adapter.SystemBus(stub_pydbus)

1752

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

1753

"objectpath", "interface",

1754

*method_args)

1755

self.assertIs(ret, expected_method_return)

1756

1757

def test_call_method_handles_exception(self):

1758

def func():

1759

raise gi.repository.GLib.Error()

1760

1761

stub_pydbus = self.Stub_pydbus_func(func)

1762

bus = pydbus_adapter.SystemBus(stub_pydbus)

1763

1764

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

1765

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

1766

"objectpath", "interface")

1767

1768

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

1769

1770

def test_get_converts_to_correct_exception(self):

1771

bus = pydbus_adapter.SystemBus(

1772

self.fake_pydbus_raises_exception_on_connect)

1773

with self.assertRaises(dbus.ConnectFailed):

1774

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

1775

"objectpath", "interface")

1776

1777

class fake_pydbus_raises_exception_on_connect:

1778

"""fake dbus-python module"""

1779

@classmethod

1780

def SystemBus(cls):

1781

def get(busname, objectpath):

1782

raise gi.repository.GLib.Error()

1783

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

1784

return Bus(get=get)

1785

1786

def test_set_property_uses_setattr(self):

1787

class Object:

1788

pass

1789

obj = Object()

1790

class pydbus_spy:

1791

class SystemBus:

1792

@staticmethod

1793

def get(busname, objectpath):

1794

return {"interface": obj}

1795

bus = pydbus_adapter.SystemBus(pydbus_spy)

1796

value = Unique()

1797

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

1798

value)

1799

self.assertIs(value, obj.key)

1800

1801

def test_get_suppresses_xml_deprecation_warning(self):

1802

if sys.version_info.major >= 3:

1803

return

1804

class stub_pydbus_get:

1805

class SystemBus:

1806

@staticmethod

1807

def get(busname, objectpath):

1808

warnings.warn_explicit(

1809

"deprecated", DeprecationWarning,

1810

"xml.etree.ElementTree", 0)

1811

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1812

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

1813

warnings.simplefilter("always")

1814

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

1815

self.assertEqual(0, len(w))

1816

1817

1818

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1819

class stub_pydbus:

1820

"""stub pydbus module"""

1821

class SystemBus:

1822

@staticmethod

1823

def get(busname, objectpath):

1824

return Unique()

1825

1826

def setUp(self):

1827

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

1828

1829

def test_returns_distinct_objectpaths(self):

1830

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

1831

self.assertIsInstance(obj1, Unique)

1832

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

1833

self.assertIsInstance(obj2, Unique)

1834

self.assertIsNot(obj1, obj2)

1835

1836

def test_returns_distinct_busnames(self):

1837

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

1838

self.assertIsInstance(obj1, Unique)

1839

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

1840

self.assertIsInstance(obj2, Unique)

1841

self.assertIsNot(obj1, obj2)

1842

1843

def test_returns_distinct_both(self):

1844

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

1845

self.assertIsInstance(obj1, Unique)

1846

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

1847

self.assertIsInstance(obj2, Unique)

1848

self.assertIsNot(obj1, obj2)

1849

1850

def test_returns_same(self):

1851

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

1852

self.assertIsInstance(obj1, Unique)

1853

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

1854

self.assertIsInstance(obj2, Unique)

1855

self.assertIs(obj1, obj2)

1856

1857

def test_returns_same_old(self):

1858

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

1859

self.assertIsInstance(obj1, Unique)

1860

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

1861

self.assertIsInstance(obj2, Unique)

1862

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

1863

self.assertIsInstance(obj1b, Unique)

1864

self.assertIsNot(obj1, obj2)

1865

self.assertIsNot(obj2, obj1b)

1866

self.assertIs(obj1, obj1b)

1867

1868

1869

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1870

1871

class dummy_dbussy:

1872

class DBUS:

1873

class ObjectPath(str):

1874

pass

1875

class DBusError(Exception):

1876

pass

1877

1878

def fake_ravel_func(self, func):

1879

class fake_ravel:

1880

@staticmethod

1881

def system_bus():

1882

class DBusInterfaceProxy:

1883

@staticmethod

1884

def methodname(*args):

1885

return [func(*args)]

1886

class DBusObject:

1887

@staticmethod

1888

def get_interface(interface):

1889

if interface == "interface":

1890

return DBusInterfaceProxy()

1891

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

1892

return fake_ravel

1893

1894

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

1895

interface, *args):

1896

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

1897

return bus.call_method(methodname, busname, objectpath,

1898

interface, *args)

1899

1900

def test_call_method_returns(self):

1901

expected_method_return = Unique()

1902

method_args = (Unique(), Unique())

1903

def func(*args):

1904

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

1905

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

1906

self.assertIs(marg, arg)

1907

return expected_method_return

1908

fake_ravel = self.fake_ravel_func(func)

1909

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

1910

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

1911

"objectpath", "interface",

1912

*method_args)

1913

self.assertIs(ret, expected_method_return)

1914

1915

def test_call_method_filters_objectpath(self):

1916

def func():

1917

return method_return

1918

fake_ravel = self.fake_ravel_func(func)

1919

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

1920

method_return = (self.dummy_dbussy.DBUS

1921

.ObjectPath("objectpath"))

1922

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

1923

"objectpath", "interface")

1924

self.assertEqual("objectpath", ret)

1925

self.assertNotIsInstance(ret,

1926

self.dummy_dbussy.DBUS.ObjectPath)

1927

1928

def test_call_method_filters_objectpaths_in_dict(self):

1929

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1930

def func():

1931

return method_return

1932

fake_ravel = self.fake_ravel_func(func)

1933

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

1934

method_return = {

1935

ObjectPath("objectpath_key_1"):

1936

ObjectPath("objectpath_value_1"),

1937

ObjectPath("objectpath_key_2"):

1938

ObjectPath("objectpath_value_2"),

1939

}

1940

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

1941

"objectpath", "interface")

1942

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

1943

for key, value in

1944

method_return.items()}

1945

for key, value in ret.items():

1946

self.assertNotIsInstance(key, ObjectPath)

1947

self.assertNotIsInstance(value, ObjectPath)

1948

self.assertEqual(expected_method_return, ret)

1949

self.assertIsInstance(ret, dict)

1950

1951

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1952

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1953

def func():

1954

return method_return

1955

fake_ravel = self.fake_ravel_func(func)

1956

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

1957

method_return = {

1958

ObjectPath("key1"): {

1959

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

1960

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

1961

},

1962

ObjectPath("key2"): {

1963

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

1964

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

1965

},

1966

}

1967

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

1968

"objectpath", "interface")

1969

expected_method_return = {

1970

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

1971

"key12": "value12"},

1972

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

1973

"key22": "value22"},

1974

}

1975

self.assertEqual(expected_method_return, ret)

1976

for key, value in ret.items():

1977

self.assertIsInstance(value, dict)

1978

self.assertEqual(expected_method_return[key], value)

1979

self.assertNotIsInstance(key, ObjectPath)

1980

for inner_key, inner_value in value.items():

1981

self.assertIsInstance(value, dict)

1982

self.assertEqual(

1983

expected_method_return[key][inner_key],

1984

inner_value)

1985

self.assertNotIsInstance(key, ObjectPath)

1986

1987

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

1988

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1989

def func():

1990

return method_return

1991

fake_ravel = self.fake_ravel_func(func)

1992

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

1993

method_return = {

1994

ObjectPath("key1"): {

1995

ObjectPath("key2"): {

1996

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

1997

},

1998

},

1999

}

2000

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

2001

"objectpath", "interface")

2002

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

2003

self.assertEqual(expected_method_return, ret)

2004

self.assertIsInstance(ret, dict)

2005

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

2006

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

2007

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

2008

ObjectPath)

2009

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

2010

self.assertNotIsInstance(

2011

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

2012

ObjectPath)

2013

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

2014

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

2015

self.dummy_dbussy.DBUS.ObjectPath)

2016

2017

def test_call_method_handles_exception(self):

2018

def func():

2019

raise self.dummy_dbussy.DBusError()

2020

2021

fake_ravel = self.fake_ravel_func(func)

2022

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

2023

2024

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

2025

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

2026

"objectpath", "interface")

2027

2028

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

2029

2030

def test_get_object_converts_to_correct_exception(self):

2031

class fake_ravel_raises_exception_on_connect:

2032

@staticmethod

2033

def system_bus():

2034

class Bus:

2035

@staticmethod

2036

def __getitem__(key):

2037

if key == "objectpath":

2038

raise self.dummy_dbussy.DBusError()

2039

raise Exception(key)

2040

return {"busname": Bus()}

2041

def func():

2042

raise self.dummy_dbussy.DBusError()

2043

bus = dbussy_adapter.SystemBus(

2044

self.dummy_dbussy,

2045

fake_ravel_raises_exception_on_connect)

2046

with self.assertRaises(dbus.ConnectFailed):

2047

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

2048

"objectpath", "interface")

2049

2050

2051

class Test_commands_from_options(unittest.TestCase):

2052

2053

def setUp(self):

2054

self.parser = argparse.ArgumentParser()

2055

add_command_line_options(self.parser)

2056

2057

def test_is_enabled(self):

2058

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

2059

command.IsEnabled)

2060

2061

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

2062

clients=None, **cmd_attrs):

2063

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

2064

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

2065

options = self.parser.parse_args(args)

2066

check_option_syntax(self.parser, options)

2067

commands = commands_from_options(options)

2068

self.assertEqual(length, len(commands))

2069

for command in commands:

2070

if isinstance(command, command_cls):

2071

break

2072

else:

2073

self.assertIsInstance(command, command_cls)

2074

if clients is not None:

2075

self.assertEqual(clients, options.client)

2076

for key, value in cmd_attrs.items():

2077

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

2078

2079

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

2080

for cmd in commands:

2081

self.assert_command_from_args(args, cmd,

2082

length=len(commands),

2083

clients=clients)

2084

2085

def test_is_enabled_short(self):

2086

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

2087

command.IsEnabled)

2088

2089

def test_approve(self):

2090

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

2091

command.Approve)

2092

2093

def test_approve_short(self):

2094

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

2095

command.Approve)

2096

2097

def test_deny(self):

2098

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

2099

command.Deny)

2100

2101

def test_deny_short(self):

2102

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

2103

2104

def test_remove(self):

2105

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

2106

command.Remove)

2107

2108

def test_deny_before_remove(self):

2109

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

2110

"client"])

2111

check_option_syntax(self.parser, options)

2112

commands = commands_from_options(options)

2113

self.assertEqual(2, len(commands))

2114

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

2115

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

2116

2117

def test_deny_before_remove_reversed(self):

2118

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

2119

"--all"])

2120

check_option_syntax(self.parser, options)

2121

commands = commands_from_options(options)

2122

self.assertEqual(2, len(commands))

2123

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

2124

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

2125

2126

def test_remove_short(self):

2127

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

2128

command.Remove)

2129

2130

def test_dump_json(self):

2131

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

2132

command.DumpJSON)

2133

2134

def test_enable(self):

2135

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

2136

command.Enable)

2137

2138

def test_enable_short(self):

2139

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

2140

command.Enable)

2141

2142

def test_disable(self):

2143

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

2144

command.Disable)

2145

2146

def test_disable_short(self):

2147

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

2148

command.Disable)

2149

2150

def test_bump_timeout(self):

2151

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

2152

command.BumpTimeout)

2153

2154

def test_bump_timeout_short(self):

2155

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

2156

command.BumpTimeout)

2157

2158

def test_start_checker(self):

2159

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

2160

command.StartChecker)

2161

2162

def test_stop_checker(self):

2163

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

2164

command.StopChecker)

2165

2166

def test_approve_by_default(self):

2167

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

2168

"client"],

2169

command.ApproveByDefault)

2170

2171

def test_deny_by_default(self):

2172

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

2173

command.DenyByDefault)

2174

2175

def test_checker(self):

2176

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

2177

command.SetChecker,

2178

value_to_set=":")

2179

2180

def test_checker_empty(self):

2181

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

2182

command.SetChecker,

2183

value_to_set="")

2184

2185

def test_checker_short(self):

2186

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

2187

command.SetChecker,

2188

value_to_set=":")

2189

2190

def test_host(self):

2191

self.assert_command_from_args(

2192

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

2193

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

2194

2195

def test_host_short(self):

2196

self.assert_command_from_args(

2197

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

2198

value_to_set="client.example.org")

2199

2200

def test_secret_devnull(self):

2201

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

2202

"client"], command.SetSecret,

2203

value_to_set=b"")

2204

2205

def test_secret_tempfile(self):

2206

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

2207

value = b"secret\0xyzzy\nbar"

2208

f.write(value)

2209

f.seek(0)

2210

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

2211

"client"],

2212

command.SetSecret,

2213

value_to_set=value)

2214

2215

def test_secret_devnull_short(self):

2216

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

2217

"client"], command.SetSecret,

2218

value_to_set=b"")

2219

2220

def test_secret_tempfile_short(self):

2221

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

2222

value = b"secret\0xyzzy\nbar"

2223

f.write(value)

2224

f.seek(0)

2225

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

2226

command.SetSecret,

2227

value_to_set=value)

2228

2229

def test_timeout(self):

2230

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

2231

command.SetTimeout,

2232

value_to_set=300000)

2233

2234

def test_timeout_short(self):

2235

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

2236

command.SetTimeout,

2237

value_to_set=300000)

2238

2239

def test_extended_timeout(self):

2240

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

2241

"client"],

2242

command.SetExtendedTimeout,

2243

value_to_set=900000)

2244

2245

def test_interval(self):

2246

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

2247

"client"], command.SetInterval,

2248

value_to_set=120000)

2249

2250

def test_interval_short(self):

2251

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

2252

command.SetInterval,

2253

value_to_set=120000)

2254

2255

def test_approval_delay(self):

2256

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

2257

"client"],

2258

command.SetApprovalDelay,

2259

value_to_set=30000)

2260

2261

def test_approval_duration(self):

2262

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

2263

"client"],

2264

command.SetApprovalDuration,

2265

value_to_set=1000)

2266

2267

def test_print_table(self):

2268

self.assert_command_from_args([], command.PrintTable,

2269

verbose=False)

2270

2271

def test_print_table_verbose(self):

2272

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

2273

command.PrintTable,

2274

verbose=True)

2275

2276

def test_print_table_verbose_short(self):

2277

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

2278

verbose=True)

2279

2280

2281

def test_manual_page_example_1(self):

2282

self.assert_command_from_args("",

2283

command.PrintTable,

2284

clients=[],

2285

verbose=False)

2286

2287

def test_manual_page_example_2(self):

2288

self.assert_command_from_args(

2289

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

2290

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

2291

"foo2.example.org"],

2292

verbose=True)

2293

2294

def test_manual_page_example_3(self):

2295

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

2296

command.Enable,

2297

clients=[])

2298

2299

def test_manual_page_example_4(self):

2300

self.assert_commands_from_args(

2301

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

2302

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

2303

[command.SetTimeout, command.SetInterval],

2304

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

2305

2306

def test_manual_page_example_5(self):

2307

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

2308

command.Approve,

2309

clients=[])

2310

2311

2312

class TestCommand(unittest.TestCase):

2313

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

2314

2315

class FakeMandosBus(dbus.MandosBus):

2316

def __init__(self, testcase):

2317

self.client_properties = {

2318

"Name": "foo",

2319

"KeyID": ("92ed150794387c03ce684574b1139a65"

2320

"94a34f895daaaf09fd8ea90a27cddb12"),

2321

"Secret": b"secret",

2322

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

2323

"Enabled": True,

2324

"Timeout": 300000,

2325

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

2326

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

2327

"Interval": 120000,

2328

"Fingerprint": ("778827225BA7DE539C5A"

2329

"7CFA59CFF7CDBD9A5920"),

2330

"CheckerRunning": False,

2331

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

2332

"ApprovalPending": False,

2333

"ApprovedByDefault": True,

2334

"LastApprovalRequest": "",

2335

"ApprovalDelay": 0,

2336

"ApprovalDuration": 1000,

2337

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

2338

"ExtendedTimeout": 900000,

2339

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

2340

"LastCheckerStatus": 0,

2341

}

2342

self.other_client_properties = {

2343

"Name": "barbar",

2344

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2345

"6ab612cff5ad227247e46c2b020f441c"),

2346

"Secret": b"secretbar",

2347

"Host": "192.0.2.3",

2348

"Enabled": True,

2349

"Timeout": 300000,

2350

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

2351

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

2352

"Interval": 120000,

2353

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2354

"F547B3A107558FCA3A27"),

2355

"CheckerRunning": True,

2356

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

2357

"ApprovalPending": False,

2358

"ApprovedByDefault": False,

2359

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

2360

"ApprovalDelay": 30000,

2361

"ApprovalDuration": 93785000,

2362

"Checker": ":",

2363

"ExtendedTimeout": 900000,

2364

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

2365

"LastCheckerStatus": -2,

2366

}

2367

self.clients = collections.OrderedDict(

2368

[

2369

("client_objectpath", self.client_properties),

2370

("other_client_objectpath",

2371

self.other_client_properties),

2372

])

2373

self.one_client = {"client_objectpath":

2374

self.client_properties}

2375

self.testcase = testcase

2376

self.calls = []

2377

2378

def call_method(self, methodname, busname, objectpath,

2379

interface, *args):

2380

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

2381

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

2382

interface, args))

2383

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

2384

if methodname == "Set":

2385

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

2386

interface, key, value = args

2387

self.testcase.assertEqual(

2388

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

2389

self.clients[objectpath][key] = value

2390

return

2391

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

2392

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

2393

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

2394

clientpath = args[0]

2395

del self.clients[clientpath]

2396

return

2397

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

2398

if methodname == "Approve":

2399

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

2400

return

2401

raise ValueError()

2402

2403

def setUp(self):

2404

self.bus = self.FakeMandosBus(self)

2405

2406

2407

class TestBaseCommands(TestCommand):

2408

2409

def test_IsEnabled_exits_successfully(self):

2410

with self.assertRaises(SystemExit) as e:

2411

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

2412

if e.exception.code is not None:

2413

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

2414

else:

2415

self.assertIsNone(e.exception.code)

2416

2417

def test_IsEnabled_exits_with_failure(self):

2418

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

2419

with self.assertRaises(SystemExit) as e:

2420

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

2421

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

2422

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

2423

else:

2424

self.assertIsNotNone(e.exception.code)

2425

2426

def test_Approve(self):

2427

busname = "se.recompile.Mandos"

2428

client_interface = "se.recompile.Mandos.Client"

2429

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

2430

self.assertTrue(self.bus.clients)

2431

for clientpath in self.bus.clients:

2432

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

2433

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

2434

2435

def test_Deny(self):

2436

busname = "se.recompile.Mandos"

2437

client_interface = "se.recompile.Mandos.Client"

2438

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

2439

self.assertTrue(self.bus.clients)

2440

for clientpath in self.bus.clients:

2441

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

2442

client_interface, (False,)),

2443

self.bus.calls)

2444

2445

def test_Remove(self):

2446

busname = "se.recompile.Mandos"

2447

server_path = "/"

2448

server_interface = "se.recompile.Mandos"

2449

orig_clients = self.bus.clients.copy()

2450

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

2451

self.assertFalse(self.bus.clients)

2452

for clientpath in orig_clients:

2453

self.assertIn(("RemoveClient", busname,

2454

server_path, server_interface,

2455

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

2456

2457

expected_json = {

2458

"foo": {

2459

"Name": "foo",

2460

"KeyID": ("92ed150794387c03ce684574b1139a65"

2461

"94a34f895daaaf09fd8ea90a27cddb12"),

2462

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

2463

"Enabled": True,

2464

"Timeout": 300000,

2465

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

2466

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

2467

"Interval": 120000,

2468

"Fingerprint": ("778827225BA7DE539C5A"

2469

"7CFA59CFF7CDBD9A5920"),

2470

"CheckerRunning": False,

2471

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

2472

"ApprovalPending": False,

2473

"ApprovedByDefault": True,

2474

"LastApprovalRequest": "",

2475

"ApprovalDelay": 0,

2476

"ApprovalDuration": 1000,

2477

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

2478

"ExtendedTimeout": 900000,

2479

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

2480

"LastCheckerStatus": 0,

2481

},

2482

"barbar": {

2483

"Name": "barbar",

2484

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2485

"6ab612cff5ad227247e46c2b020f441c"),

2486

"Host": "192.0.2.3",

2487

"Enabled": True,

2488

"Timeout": 300000,

2489

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

2490

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

2491

"Interval": 120000,

2492

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2493

"F547B3A107558FCA3A27"),

2494

"CheckerRunning": True,

2495

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

2496

"ApprovalPending": False,

2497

"ApprovedByDefault": False,

2498

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

2499

"ApprovalDelay": 30000,

2500

"ApprovalDuration": 93785000,

2501

"Checker": ":",

2502

"ExtendedTimeout": 900000,

2503

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

2504

"LastCheckerStatus": -2,

2505

},

2506

}

2507

2508

def test_DumpJSON_normal(self):

2509

with self.capture_stdout_to_buffer() as buffer:

2510

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

2511

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

2512

self.assertDictEqual(self.expected_json, json_data)

2513

2514

@staticmethod

2515

@contextlib.contextmanager

2516

def capture_stdout_to_buffer():

2517

capture_buffer = io.StringIO()

2518

old_stdout = sys.stdout

2519

sys.stdout = capture_buffer

2520

try:

2521

yield capture_buffer

2522

finally:

2523

sys.stdout = old_stdout

2524

2525

def test_DumpJSON_one_client(self):

2526

with self.capture_stdout_to_buffer() as buffer:

2527

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

2528

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

2529

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

2530

self.assertDictEqual(expected_json, json_data)

2531

2532

def test_PrintTable_normal(self):

2533

with self.capture_stdout_to_buffer() as buffer:

2534

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

2535

expected_output = "\n".join((

2536

"Name Enabled Timeout Last Successful Check",

2537

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

2538

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

2539

)) + "\n"

2540

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

2541

2542

def test_PrintTable_verbose(self):

2543

with self.capture_stdout_to_buffer() as buffer:

2544

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

2545

columns = (

2546

(

2547

"Name ",

2548

"foo ",

2549

"barbar ",

2550

),(

2551

"Enabled ",

2552

"Yes ",

2553

"Yes ",

2554

),(

2555

"Timeout ",

2556

"00:05:00 ",

2557

"00:05:00 ",

2558

),(

2559

"Last Successful Check ",

2560

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

2561

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

2562

),(

2563

"Created ",

2564

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

2565

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

2566

),(

2567

"Interval ",

2568

"00:02:00 ",

2569

"00:02:00 ",

2570

),(

2571

"Host ",

2572

"foo.example.org ",

2573

"192.0.2.3 ",

2574

),(

2575

("Key ID "

2576

" "),

2577

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2578

"ea90a27cddb12 "),

2579

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2580

"46c2b020f441c "),

2581

),(

2582

"Fingerprint ",

2583

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2584

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2585

),(

2586

"Check Is Running ",

2587

"No ",

2588

"Yes ",

2589

),(

2590

"Last Enabled ",

2591

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

2592

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

2593

),(

2594

"Approval Is Pending ",

2595

"No ",

2596

"No ",

2597

),(

2598

"Approved By Default ",

2599

"Yes ",

2600

"No ",

2601

),(

2602

"Last Approval Request ",

2603

" ",

2604

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

2605

),(

2606

"Approval Delay ",

2607

"00:00:00 ",

2608

"00:00:30 ",

2609

),(

2610

"Approval Duration ",

2611

"00:00:01 ",

2612

"1T02:03:05 ",

2613

),(

2614

"Checker ",

2615

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

2616

": ",

2617

),(

2618

"Extended Timeout ",

2619

"00:15:00 ",

2620

"00:15:00 ",

2621

),(

2622

"Expires ",

2623

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

2624

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

2625

),(

2626

"Last Checker Status",

2627

"0 ",

2628

"-2 ",

2629

)

2630

)

2631

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

2632

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

2633

for rows in columns)

2634

for line in range(num_lines))

2635

+ "\n")

2636

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

2637

2638

def test_PrintTable_one_client(self):

2639

with self.capture_stdout_to_buffer() as buffer:

2640

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

2641

expected_output = "\n".join((

2642

"Name Enabled Timeout Last Successful Check",

2643

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

2644

)) + "\n"

2645

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

2646

2647

2648

class TestPropertySetterCmd(TestCommand):

2649

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

2650

2651

def runTest(self):

2652

if not hasattr(self, "command"):

2653

return # Abstract TestCase class

2654

2655

if hasattr(self, "values_to_set"):

2656

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

2657

values_to_get = getattr(self, "values_to_get",

2658

self.values_to_set)

2659

else:

2660

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

2661

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2663

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

2664

for clientpath in self.bus.clients:

2665

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

2666

Unique())

2667

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

2668

self.assertTrue(self.bus.clients)

2669

for clientpath in self.bus.clients:

2670

value = (self.bus.clients[clientpath]

2671

[self.propname])

2672

self.assertNotIsInstance(value, Unique)

2673

self.assertEqual(value_to_get, value)

2674

2675

2676

class TestEnableCmd(TestPropertySetterCmd):

2677

command = command.Enable

2678

propname = "Enabled"

2679

values_to_get = [True]

2680

2681

2682

class TestDisableCmd(TestPropertySetterCmd):

2683

command = command.Disable

2684

propname = "Enabled"

2685

values_to_get = [False]

2686

2687

2688

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2689

command = command.BumpTimeout

2690

propname = "LastCheckedOK"

2691

values_to_get = [""]

2692

2693

2694

class TestStartCheckerCmd(TestPropertySetterCmd):

2695

command = command.StartChecker

2696

propname = "CheckerRunning"

2697

values_to_get = [True]

2698

2699

2700

class TestStopCheckerCmd(TestPropertySetterCmd):

2701

command = command.StopChecker

2702

propname = "CheckerRunning"

2703

values_to_get = [False]

2704

2705

2706

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2707

command = command.ApproveByDefault

2708

propname = "ApprovedByDefault"

2709

values_to_get = [True]

2710

2711

2712

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2713

command = command.DenyByDefault

2714

propname = "ApprovedByDefault"

2715

values_to_get = [False]

2716

2717

2718

class TestSetCheckerCmd(TestPropertySetterCmd):

2719

command = command.SetChecker

2720

propname = "Checker"

2721

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

2722

2723

2724

class TestSetHostCmd(TestPropertySetterCmd):

2725

command = command.SetHost

2726

propname = "Host"

2727

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

2728

2729

2730

class TestSetSecretCmd(TestPropertySetterCmd):

2731

command = command.SetSecret

2732

propname = "Secret"

2733

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

2734

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

2735

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

2736

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

2737

self.values_to_set]

2738

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

2739

2740

2741

class TestSetTimeoutCmd(TestPropertySetterCmd):

2742

command = command.SetTimeout

2743

propname = "Timeout"

2744

values_to_set = [datetime.timedelta(),

2745

datetime.timedelta(minutes=5),

2746

datetime.timedelta(seconds=1),

2747

datetime.timedelta(weeks=1),

2748

datetime.timedelta(weeks=52)]

2749

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

2750

2751

2752

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2753

command = command.SetExtendedTimeout

2754

propname = "ExtendedTimeout"

2755

values_to_set = [datetime.timedelta(),

2756

datetime.timedelta(minutes=5),

2757

datetime.timedelta(seconds=1),

2758

datetime.timedelta(weeks=1),

2759

datetime.timedelta(weeks=52)]

2760

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

2761

2762

2763

class TestSetIntervalCmd(TestPropertySetterCmd):

2764

command = command.SetInterval

2765

propname = "Interval"

2766

values_to_set = [datetime.timedelta(),

2767

datetime.timedelta(minutes=5),

2768

datetime.timedelta(seconds=1),

2769

datetime.timedelta(weeks=1),

2770

datetime.timedelta(weeks=52)]

2771

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

2772

2773

2774

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2775

command = command.SetApprovalDelay

2776

propname = "ApprovalDelay"

2777

values_to_set = [datetime.timedelta(),

2778

datetime.timedelta(minutes=5),

2779

datetime.timedelta(seconds=1),

2780

datetime.timedelta(weeks=1),

2781

datetime.timedelta(weeks=52)]

2782

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

2783

2784

2785

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2786

command = command.SetApprovalDuration

2787

propname = "ApprovalDuration"

2788

values_to_set = [datetime.timedelta(),

2789

datetime.timedelta(minutes=5),

2790

datetime.timedelta(seconds=1),

2791

datetime.timedelta(weeks=1),

2792

datetime.timedelta(weeks=52)]

2793

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

2794

2795

2796

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2799

parser = argparse.ArgumentParser(add_help=False)

2800

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

2801

parser.add_argument("--prefix", )

2802

args, unknown_args = parser.parse_known_args()

2803

if args.check:

2804

# Remove test options from sys.argv

2805

sys.argv[1:] = unknown_args

2806

return args

2807

2808

# Add all tests from doctest strings

2809

def load_tests(loader, tests, none):

2810

import doctest

2811

tests.addTests(doctest.DocTestSuite())

2812

return tests

2813

2814

if __name__ == "__main__":

2815

options = parse_test_args()

2816

try:

2817

if options.check:

2818

extra_test_prefix = options.prefix

2819

if extra_test_prefix is not None:

2820

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

2821

.result.wasSuccessful()):

2822

sys.exit(1)

2823

class ExtraTestLoader(unittest.TestLoader):

2824

testMethodPrefix = extra_test_prefix

2825

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

2826

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

2827

else:

2828

unittest.main(argv=[""])

2829

else:

2830

main()

2831

finally:

2832

logging.shutdown()

2833

2834

# Local Variables:

2835

# run-tests:

2836

# (lambda (&optional extra)

2837

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

2838

# extra))

2839

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

2840

# (funcall remove-test-window)

2841

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

2842

# run-tests-in-test-buffer:

2843

# (lambda (dir &optional extra)

2844

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

2845

# (setq buffer-read-only nil

2846

# default-directory dir)

2847

# (erase-buffer)

2848

# (compilation-mode))

2849

# (let ((process-result

2850

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

2851

# (process-file-shell-command

2852

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

2853

# (and (numberp process-result)

2854

# (= process-result 0))))

2855

# get-command-line:

2856

# (lambda (&optional extra)

2857

# (let ((quoted-script

2858

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

2859

# (format

2860

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

2861

# quoted-script)))

2862

# get-script-name:

2863

# (lambda ()

2864

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

2865

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

2866

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

2867

# (buffer-file-name))))

2868

# remove-test-window:

2869

# (lambda ()

2870

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

2871

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

2872

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

2873

# (lambda ()

2874

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

2875

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

2876

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

2877

# (display-buffer-overriding-action

2878

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

2879

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

2880

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

2881

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

2882

# eval:

2883

# (progn

2884

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

2885

# (funcall run-tests t)))

2886

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

2887

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

2888

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

2889

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

2890

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

2891

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

2892

# End: