/mandos/trunk : revision 1097

1

#!/usr/bin/python

2

# -*- mode: python; coding: utf-8 -*-

3

4

import dbus

5

from optparse import OptionParser

2

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

3

#

4

# Mandos Monitor - Control and monitor 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

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

35

import sys

36

import argparse

6

37

import locale

7

8

locale.setlocale(locale.LC_ALL, u'')

9

10

tablewords = {

11

'name': u'Name',

12

'enabled': u'Enabled',

13

'timeout': u'Timeout',

14

'last_checked_ok': u'Last Successful Check',

15

'created': u'Created',

16

'interval': u'Interval',

17

'host': u'Host',

18

'fingerprint': u'Fingerprint',

19

'checker_running': u'Check Is Running',

20

'last_enabled': u'Last Enabled',

21

'checker': u'Checker',

22

}

23

busname = 'org.mandos-system.Mandos'

24

object_path = '/Mandos'

25

interface = 'org.mandos_system.Mandos'

26

version = "1.0.2"

27

defaultkeywords = ('name', 'enabled', 'timeout', 'last_checked_ok',

28

'checker')

29

30

parser = OptionParser(version = "%%prog %s" % version)

31

parser.add_option("-a", "--all", action="store_true", default=False,

32

help="Print all fields")

33

options = parser.parse_args()[0]

34

if options.all:

35

keywords = ('name', 'enabled', 'timeout', 'last_checked_ok',

36

'created', 'interval', 'host', 'fingerprint',

37

'checker_running', 'last_enabled', 'checker')

38

else:

39

keywords = defaultkeywords

40

41

42

bus = dbus.SystemBus()

43

mandos_dbus_objc = bus.get_object(busname, object_path)

44

mandos_serv = dbus.Interface(mandos_dbus_objc,

45

dbus_interface = interface)

46

mandos_clients = mandos_serv.GetAllClientsWithProperties()

47

48

def valuetostring(x):

49

if type(x) is dbus.Boolean:

50

return u"Yes" if x else u"No"

51

else:

52

return unicode(x)

53

54

format_string = u' '.join(u'%%-%ds'

55

% max(len(tablewords[key]),

56

max(len(valuetostring(client[key]))

57

for client

58

in mandos_clients.itervalues()))

59

for key in keywords)

60

print format_string % tuple(tablewords[key] for key in keywords)

61

for client in mandos_clients.itervalues():

62

print format_string % tuple(valuetostring(client[key])

63

for key in keywords)

64

65

38

import datetime

39

import re

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

try:

50

import pydbus

51

import gi

52

dbus_python = None

53

except ImportError:

54

import dbus as dbus_python

55

pydbus = None

56

class gi(object):

57

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

58

class repository(object):

59

class GLib(object):

60

class Error(Exception):

61

pass

62

63

# Show warnings by default

64

if not sys.warnoptions:

65

import warnings

66

warnings.simplefilter("default")

67

68

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

69

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

70

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

71

72

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

73

74

if sys.version_info.major == 2:

75

str = unicode

76

import StringIO

77

io.StringIO = StringIO.StringIO

78

79

locale.setlocale(locale.LC_ALL, "")

80

81

version = "1.8.3"

82

83

84

def main():

85

parser = argparse.ArgumentParser()

86

add_command_line_options(parser)

87

88

options = parser.parse_args()

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

if pydbus is not None:

97

bus = pydbus_adapter.CachingBus(pydbus)

98

else:

99

bus = dbus_python_adapter.CachingBus(dbus_python)

100

101

try:

102

all_clients = bus.get_clients_and_properties()

103

except dbus.ConnectFailed as e:

104

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

105

sys.exit(1)

106

except dbus.Error as e:

107

log.critical(

108

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

109

sys.exit(1)

110

111

# Compile dict of (clientpath: properties) to process

112

if not clientnames:

113

clients = all_clients

114

else:

115

clients = {}

116

for name in clientnames:

117

for objpath, properties in all_clients.items():

118

if properties["Name"] == name:

119

clients[objpath] = properties

120

break

121

else:

122

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

123

sys.exit(1)

124

125

commands = commands_from_options(options)

126

127

for command in commands:

128

command.run(clients, bus)

129

130

131

def add_command_line_options(parser):

132

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

133

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

134

help="show version number and exit")

135

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

136

help="Select all clients")

137

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

138

help="Print all fields")

139

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

140

action="append_const", default=[],

141

const=command.DumpJSON(),

142

help="Dump client data in JSON format")

143

enable_disable = parser.add_mutually_exclusive_group()

144

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

145

action="append_const", default=[],

146

const=command.Enable(),

147

help="Enable client")

148

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

149

action="append_const", default=[],

150

const=command.Disable(),

151

help="disable client")

152

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

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

155

help="Bump timeout for client")

156

start_stop_checker = parser.add_mutually_exclusive_group()

157

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

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

161

help="Start checker for client")

162

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

163

action="append_const", default=[],

164

const=command.StopChecker(),

165

help="Stop checker for client")

166

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

167

action="append_const", default=[],

168

const=command.IsEnabled(),

169

help="Check if client is enabled")

170

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

171

action="append_const", default=[],

172

const=command.Remove(),

173

help="Remove client")

174

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

175

action="append", default=[],

176

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

177

help="Set checker command for client")

178

parser.add_argument(

179

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

180

default=[], metavar="TIME",

181

type=command.SetTimeout.argparse(string_to_delta),

182

help="Set timeout for client")

183

parser.add_argument(

184

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

185

default=[], metavar="TIME",

186

type=command.SetExtendedTimeout.argparse(string_to_delta),

187

help="Set extended timeout for client")

188

parser.add_argument(

189

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

190

default=[], metavar="TIME",

191

type=command.SetInterval.argparse(string_to_delta),

192

help="Set checker interval for client")

193

approve_deny_default = parser.add_mutually_exclusive_group()

194

approve_deny_default.add_argument(

195

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

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

198

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

199

approve_deny_default.add_argument(

200

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

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

203

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

204

parser.add_argument(

205

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

206

default=[], metavar="TIME",

207

type=command.SetApprovalDelay.argparse(string_to_delta),

208

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

209

parser.add_argument(

210

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

211

default=[], metavar="TIME",

212

type=command.SetApprovalDuration.argparse(string_to_delta),

213

help="Set duration of one client approval")

214

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

215

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

216

type=command.SetHost,

217

help="Set host for client")

218

parser.add_argument(

219

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

220

default=[], metavar="FILENAME",

221

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

222

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

223

approve_deny = parser.add_mutually_exclusive_group()

224

approve_deny.add_argument(

225

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

226

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

227

help="Approve any current client request")

228

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

229

action="append_const", default=[],

230

const=command.Deny(),

231

help="Deny any current client request")

232

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

233

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

234

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

235

help="Run self-test")

236

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

237

238

239

def string_to_delta(interval):

240

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

241

242

try:

243

return rfc3339_duration_to_delta(interval)

244

except ValueError as e:

245

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

246

' '.join(e.args))

247

return parse_pre_1_6_1_interval(interval)

248

249

250

def rfc3339_duration_to_delta(duration):

251

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

252

253

>>> rfc3339_duration_to_delta("P7D")

254

datetime.timedelta(7)

255

>>> rfc3339_duration_to_delta("PT60S")

256

datetime.timedelta(0, 60)

257

>>> rfc3339_duration_to_delta("PT60M")

258

datetime.timedelta(0, 3600)

259

>>> rfc3339_duration_to_delta("P60M")

260

datetime.timedelta(1680)

261

>>> rfc3339_duration_to_delta("PT24H")

262

datetime.timedelta(1)

263

>>> rfc3339_duration_to_delta("P1W")

264

datetime.timedelta(7)

265

>>> rfc3339_duration_to_delta("PT5M30S")

266

datetime.timedelta(0, 330)

267

>>> rfc3339_duration_to_delta("P1DT3M20S")

268

datetime.timedelta(1, 200)

269

>>> # Can not be empty:

270

>>> rfc3339_duration_to_delta("")

271

Traceback (most recent call last):

272

...

273

ValueError: Invalid RFC 3339 duration: ""

274

>>> # Must start with "P":

275

>>> rfc3339_duration_to_delta("1D")

276

Traceback (most recent call last):

277

...

278

ValueError: Invalid RFC 3339 duration: "1D"

279

>>> # Must use correct order

280

>>> rfc3339_duration_to_delta("PT1S2M")

281

Traceback (most recent call last):

282

...

283

ValueError: Invalid RFC 3339 duration: "PT1S2M"

284

>>> # Time needs time marker

285

>>> rfc3339_duration_to_delta("P1H2S")

286

Traceback (most recent call last):

287

...

288

ValueError: Invalid RFC 3339 duration: "P1H2S"

289

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

290

>>> rfc3339_duration_to_delta("P1D2W")

291

Traceback (most recent call last):

292

...

293

ValueError: Invalid RFC 3339 duration: "P1D2W"

294

>>> rfc3339_duration_to_delta("P2W2H")

295

Traceback (most recent call last):

296

...

297

ValueError: Invalid RFC 3339 duration: "P2W2H"

298

"""

299

300

# Parsing an RFC 3339 duration with regular expressions is not

301

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

302

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

303

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

304

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

305

# avoid excessive use of external libraries.

306

307

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

308

Token = collections.namedtuple("Token", (

309

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

310

# a "group" containing digits

311

"value", # datetime.timedelta or None

312

"followers")) # Tokens valid after this token

313

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

314

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

315

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

316

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

317

datetime.timedelta(seconds=1),

318

frozenset((token_end, )))

319

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

320

datetime.timedelta(minutes=1),

321

frozenset((token_second, token_end)))

322

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

323

datetime.timedelta(hours=1),

324

frozenset((token_minute, token_end)))

325

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

326

None,

327

frozenset((token_hour, token_minute,

328

token_second)))

329

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

330

datetime.timedelta(days=1),

331

frozenset((token_time, token_end)))

332

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

333

datetime.timedelta(weeks=4),

334

frozenset((token_day, token_end)))

335

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

336

datetime.timedelta(weeks=52),

337

frozenset((token_month, token_end)))

338

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

339

datetime.timedelta(weeks=1),

340

frozenset((token_end, )))

341

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

342

frozenset((token_year, token_month,

343

token_day, token_time,

344

token_week)))

345

# Define starting values:

346

# Value so far

347

value = datetime.timedelta()

348

found_token = None

349

# Following valid tokens

350

followers = frozenset((token_duration, ))

351

# String left to parse

352

s = duration

353

# Loop until end token is found

354

while found_token is not token_end:

355

# Search for any currently valid tokens

356

for token in followers:

357

match = token.regexp.match(s)

358

if match is not None:

359

# Token found

360

if token.value is not None:

361

# Value found, parse digits

362

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

363

# Add to value so far

364

value += factor * token.value

365

# Strip token from string

366

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

367

# Go to found token

368

found_token = token

369

# Set valid next tokens

370

followers = found_token.followers

371

break

372

else:

373

# No currently valid tokens were found

374

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

375

.format(duration))

376

# End token found

377

return value

378

379

380

def parse_pre_1_6_1_interval(interval):

381

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

382

and return a datetime.timedelta

383

384

>>> parse_pre_1_6_1_interval('7d')

385

datetime.timedelta(7)

386

>>> parse_pre_1_6_1_interval('60s')

387

datetime.timedelta(0, 60)

388

>>> parse_pre_1_6_1_interval('60m')

389

datetime.timedelta(0, 3600)

390

>>> parse_pre_1_6_1_interval('24h')

391

datetime.timedelta(1)

392

>>> parse_pre_1_6_1_interval('1w')

393

datetime.timedelta(7)

394

>>> parse_pre_1_6_1_interval('5m 30s')

395

datetime.timedelta(0, 330)

396

>>> parse_pre_1_6_1_interval('')

397

datetime.timedelta(0)

398

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

399

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

400

datetime.timedelta(2, 480, 18000)

401

402

"""

403

404

value = datetime.timedelta(0)

405

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

406

407

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

408

if suffix == "d":

409

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

410

elif suffix == "s":

411

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

412

elif suffix == "m":

413

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

414

elif suffix == "h":

415

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

416

elif suffix == "w":

417

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

418

elif suffix == "":

419

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

420

return value

421

422

423

def check_option_syntax(parser, options):

424

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

425

argparse"""

426

427

def has_commands(options, commands=None):

428

if commands is None:

429

commands = (command.Enable,

430

command.Disable,

431

command.BumpTimeout,

432

command.StartChecker,

433

command.StopChecker,

434

command.IsEnabled,

435

command.Remove,

436

command.SetChecker,

437

command.SetTimeout,

438

command.SetExtendedTimeout,

439

command.SetInterval,

440

command.ApproveByDefault,

441

command.DenyByDefault,

442

command.SetApprovalDelay,

443

command.SetApprovalDuration,

444

command.SetHost,

445

command.SetSecret,

446

command.Approve,

447

command.Deny)

448

return any(isinstance(cmd, commands)

449

for cmd in options.commands)

450

451

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

452

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

453

if options.verbose and has_commands(options):

454

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

455

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

456

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

457

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

458

if options.all and not has_commands(options):

459

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

460

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

461

and len(options.client) > 1):

462

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

463

if (len(options.commands) > 1

464

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

465

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

466

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

467

468

469

class dbus(object):

470

471

class SystemBus(object):

472

473

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

474

def get_managed_objects(self, busname, objectpath):

475

return self.call_method("GetManagedObjects", busname,

476

objectpath,

477

self.object_manager_iface)

478

479

properties_iface = "org.freedesktop.DBus.Properties"

480

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

481

value):

482

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

483

self.properties_iface, interface, key,

484

value)

485

486

487

class MandosBus(SystemBus):

488

busname_domain = "se.recompile"

489

busname = busname_domain + ".Mandos"

490

server_path = "/"

491

server_interface = busname_domain + ".Mandos"

492

client_interface = busname_domain + ".Mandos.Client"

493

del busname_domain

494

495

def get_clients_and_properties(self):

496

managed_objects = self.get_managed_objects(

497

self.busname, self.server_path)

498

return {objpath: properties[self.client_interface]

499

for objpath, properties in managed_objects.items()

500

if self.client_interface in properties}

501

502

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

503

return self.set_property(self.busname, objectpath,

504

self.client_interface, key,

505

value)

506

507

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

508

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

509

self.client_interface, *args)

510

511

def call_server_method(self, method, *args):

512

return self.call_method(method, self.busname,

513

self.server_path,

514

self.server_interface, *args)

515

516

class Error(Exception):

517

pass

518

519

class ConnectFailed(Error):

520

pass

521

522

523

class dbus_python_adapter(object):

524

525

class SystemBus(dbus.MandosBus):

526

"""Use dbus-python"""

527

528

def __init__(self, module=dbus_python):

529

self.dbus_python = module

530

self.bus = self.dbus_python.SystemBus()

531

532

@contextlib.contextmanager

533

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

534

try:

535

yield

536

except self.dbus_python.exceptions.DBusException as e:

537

# This does what "raise from" would do

538

exc = exception_class(*e.args)

539

exc.__cause__ = e

540

raise exc

541

542

def call_method(self, methodname, busname, objectpath,

543

interface, *args):

544

proxy_object = self.get_object(busname, objectpath)

545

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

546

interface, methodname,

547

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

548

method = getattr(proxy_object, methodname)

549

with self.convert_exception():

550

with dbus_python_adapter.SilenceLogger(

551

"dbus.proxies"):

552

value = method(*args, dbus_interface=interface)

553

return self.type_filter(value)

554

555

def get_object(self, busname, objectpath):

556

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

557

busname, objectpath)

558

with self.convert_exception(dbus.ConnectFailed):

559

return self.bus.get_object(busname, objectpath)

560

561

def type_filter(self, value):

562

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

563

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

564

return bool(value)

565

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

566

return str(value)

567

# Also recurse into dictionaries

568

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

569

return {self.type_filter(key):

570

self.type_filter(subval)

571

for key, subval in value.items()}

572

return value

573

574

575

class SilenceLogger(object):

576

"Simple context manager to silence a particular logger"

577

def __init__(self, loggername):

578

self.logger = logging.getLogger(loggername)

579

580

def __enter__(self):

581

self.logger.addFilter(self.nullfilter)

582

583

class NullFilter(logging.Filter):

584

def filter(self, record):

585

return False

586

587

nullfilter = NullFilter()

588

589

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

590

self.logger.removeFilter(self.nullfilter)

591

592

593

class CachingBus(SystemBus):

594

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

595

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

596

self.object_cache = {}

597

super(dbus_python_adapter.CachingBus,

598

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

599

def get_object(self, busname, objectpath):

600

try:

601

return self.object_cache[(busname, objectpath)]

602

except KeyError:

603

new_object = super(

604

dbus_python_adapter.CachingBus,

605

self).get_object(busname, objectpath)

606

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

607

return new_object

608

609

610

class pydbus_adapter(object):

611

class SystemBus(dbus.MandosBus):

612

def __init__(self, module=pydbus):

613

self.pydbus = module

614

self.bus = self.pydbus.SystemBus()

615

616

@contextlib.contextmanager

617

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

618

try:

619

yield

620

except gi.repository.GLib.Error as e:

621

# This does what "raise from" would do

622

exc = exception_class(*e.args)

623

exc.__cause__ = e

624

raise exc

625

626

def call_method(self, methodname, busname, objectpath,

627

interface, *args):

628

proxy_object = self.get(busname, objectpath)

629

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

630

interface, methodname,

631

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

632

method = getattr(proxy_object[interface], methodname)

633

with self.convert_exception():

634

return method(*args)

635

636

def get(self, busname, objectpath):

637

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

638

busname, objectpath)

639

with self.convert_exception(dbus.ConnectFailed):

640

if sys.version_info.major <= 2:

641

with warnings.catch_warnings():

642

warnings.filterwarnings(

643

"ignore", "", DeprecationWarning,

644

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

645

return self.bus.get(busname, objectpath)

646

else:

647

return self.bus.get(busname, objectpath)

648

649

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

650

value):

651

proxy_object = self.get(busname, objectpath)

652

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

653

objectpath, self.properties_iface, interface,

654

key, value)

655

setattr(proxy_object[interface], key, value)

656

657

class CachingBus(SystemBus):

658

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

659

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

660

self.object_cache = {}

661

super(pydbus_adapter.CachingBus,

662

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

663

def get(self, busname, objectpath):

664

try:

665

return self.object_cache[(busname, objectpath)]

666

except KeyError:

667

new_object = (super(pydbus_adapter.CachingBus, self)

668

.get(busname, objectpath))

669

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

670

return new_object

671

672

673

def commands_from_options(options):

674

675

commands = list(options.commands)

676

677

def find_cmd(cmd, commands):

678

i = 0

679

for i, c in enumerate(commands):

680

if isinstance(c, cmd):

681

return i

682

return i+1

683

684

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

685

# to occur ahead of command.Remove.

686

index_of_remove = find_cmd(command.Remove, commands)

687

before_remove = commands[:index_of_remove]

688

after_remove = commands[index_of_remove:]

689

cleaned_after = []

690

for cmd in after_remove:

691

if isinstance(cmd, command.Deny):

692

before_remove.append(cmd)

693

else:

694

cleaned_after.append(cmd)

695

if cleaned_after != after_remove:

696

commands = before_remove + cleaned_after

697

698

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

699

# optionally verbosely

700

if not commands:

701

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

702

703

return commands

704

705

706

class command(object):

707

"""A namespace for command classes"""

708

709

class Base(object):

710

"""Abstract base class for commands"""

711

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

712

"""Normal commands should implement run_on_one_client(),

713

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

714

override this run() method instead.

715

"""

716

self.bus = bus

717

for client, properties in clients.items():

718

self.run_on_one_client(client, properties)

719

720

721

class IsEnabled(Base):

722

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

723

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

724

if properties["Enabled"]:

725

sys.exit(0)

726

sys.exit(1)

727

728

729

class Approve(Base):

730

def run_on_one_client(self, client, properties):

731

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

732

733

734

class Deny(Base):

735

def run_on_one_client(self, client, properties):

736

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

737

738

739

class Remove(Base):

740

def run(self, clients, bus):

741

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

742

bus.call_server_method("RemoveClient", clientpath)

743

744

745

class Output(Base):

746

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

747

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

748

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

749

"Fingerprint", "CheckerRunning",

750

"LastEnabled", "ApprovalPending",

751

"ApprovedByDefault", "LastApprovalRequest",

752

"ApprovalDelay", "ApprovalDuration",

753

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

754

"LastCheckerStatus")

755

756

757

class DumpJSON(Output):

758

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

759

data = {properties["Name"]:

760

{key: properties[key]

761

for key in self.all_keywords}

762

for properties in clients.values()}

763

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

764

765

766

class PrintTable(Output):

767

def __init__(self, verbose=False):

768

self.verbose = verbose

769

770

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

771

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

772

"LastCheckedOK")

773

keywords = default_keywords

774

if self.verbose:

775

keywords = self.all_keywords

776

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

777

778

class TableOfClients(object):

779

tableheaders = {

780

"Name": "Name",

781

"Enabled": "Enabled",

782

"Timeout": "Timeout",

783

"LastCheckedOK": "Last Successful Check",

784

"LastApprovalRequest": "Last Approval Request",

785

"Created": "Created",

786

"Interval": "Interval",

787

"Host": "Host",

788

"Fingerprint": "Fingerprint",

789

"KeyID": "Key ID",

790

"CheckerRunning": "Check Is Running",

791

"LastEnabled": "Last Enabled",

792

"ApprovalPending": "Approval Is Pending",

793

"ApprovedByDefault": "Approved By Default",

794

"ApprovalDelay": "Approval Delay",

795

"ApprovalDuration": "Approval Duration",

796

"Checker": "Checker",

797

"ExtendedTimeout": "Extended Timeout",

798

"Expires": "Expires",

799

"LastCheckerStatus": "Last Checker Status",

800

}

801

802

def __init__(self, clients, keywords):

803

self.clients = clients

804

self.keywords = keywords

805

806

def __str__(self):

807

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

808

809

if sys.version_info.major == 2:

810

__unicode__ = __str__

811

def __str__(self):

812

return str(self).encode(

813

locale.getpreferredencoding())

814

815

def rows(self):

816

format_string = self.row_formatting_string()

817

rows = [self.header_line(format_string)]

818

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

819

for client in self.clients)

820

return rows

821

822

def row_formatting_string(self):

823

"Format string used to format table rows"

824

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

825

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

826

*(len(self.string_from_client(client,

827

key))

828

for client in self.clients)),

829

key=key)

830

for key in self.keywords)

831

832

def string_from_client(self, client, key):

833

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

834

835

@classmethod

836

def valuetostring(cls, value, keyword):

837

if isinstance(value, bool):

838

return "Yes" if value else "No"

839

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

840

"ApprovalDuration", "ExtendedTimeout"):

841

return cls.milliseconds_to_string(value)

842

return str(value)

843

844

def header_line(self, format_string):

845

return format_string.format(**self.tableheaders)

846

847

def client_line(self, client, format_string):

848

return format_string.format(

849

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

850

for key in self.keywords})

851

852

@staticmethod

853

def milliseconds_to_string(ms):

854

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

855

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

856

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

857

if td.days else "",

858

hours=td.seconds // 3600,

859

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

860

seconds=td.seconds % 60))

861

862

863

class PropertySetter(Base):

864

"Abstract class for Actions for setting one client property"

865

866

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

867

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

868

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

869

self.value_to_set)

870

871

@property

872

def propname(self):

873

raise NotImplementedError()

874

875

876

class Enable(PropertySetter):

877

propname = "Enabled"

878

value_to_set = True

879

880

881

class Disable(PropertySetter):

882

propname = "Enabled"

883

value_to_set = False

884

885

886

class BumpTimeout(PropertySetter):

887

propname = "LastCheckedOK"

888

value_to_set = ""

889

890

891

class StartChecker(PropertySetter):

892

propname = "CheckerRunning"

893

value_to_set = True

894

895

896

class StopChecker(PropertySetter):

897

propname = "CheckerRunning"

898

value_to_set = False

899

900

901

class ApproveByDefault(PropertySetter):

902

propname = "ApprovedByDefault"

903

value_to_set = True

904

905

906

class DenyByDefault(PropertySetter):

907

propname = "ApprovedByDefault"

908

value_to_set = False

909

910

911

class PropertySetterValue(PropertySetter):

912

"""Abstract class for PropertySetter recieving a value as

913

constructor argument instead of a class attribute."""

914

def __init__(self, value):

915

self.value_to_set = value

916

917

@classmethod

918

def argparse(cls, argtype):

919

def cmdtype(arg):

920

return cls(argtype(arg))

921

return cmdtype

922

923

class SetChecker(PropertySetterValue):

924

propname = "Checker"

925

926

927

class SetHost(PropertySetterValue):

928

propname = "Host"

929

930

931

class SetSecret(PropertySetterValue):

932

propname = "Secret"

933

934

@property

935

def value_to_set(self):

936

return self._vts

937

938

@value_to_set.setter

939

def value_to_set(self, value):

940

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

941

self._vts = value.read()

942

value.close()

943

944

945

class PropertySetterValueMilliseconds(PropertySetterValue):

946

"""Abstract class for PropertySetterValue taking a value

947

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

948

milliseconds."""

949

950

@property

951

def value_to_set(self):

952

return self._vts

953

954

@value_to_set.setter

955

def value_to_set(self, value):

956

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

957

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

958

959

960

class SetTimeout(PropertySetterValueMilliseconds):

961

propname = "Timeout"

962

963

964

class SetExtendedTimeout(PropertySetterValueMilliseconds):

965

propname = "ExtendedTimeout"

966

967

968

class SetInterval(PropertySetterValueMilliseconds):

969

propname = "Interval"

970

971

972

class SetApprovalDelay(PropertySetterValueMilliseconds):

973

propname = "ApprovalDelay"

974

975

976

class SetApprovalDuration(PropertySetterValueMilliseconds):

977

propname = "ApprovalDuration"

978

979

980

981

class TestCaseWithAssertLogs(unittest.TestCase):

982

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

983

984

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

985

@contextlib.contextmanager

986

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

987

capturing_handler = self.CapturingLevelHandler(level)

988

old_level = logger.level

989

old_propagate = logger.propagate

990

logger.addHandler(capturing_handler)

991

logger.setLevel(level)

992

logger.propagate = False

993

try:

994

yield capturing_handler.watcher

995

finally:

996

logger.propagate = old_propagate

997

logger.removeHandler(capturing_handler)

998

logger.setLevel(old_level)

999

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

1000

0)

1001

1002

class CapturingLevelHandler(logging.Handler):

1003

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

1004

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

1005

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

1006

def emit(self, record):

1007

self.watcher.records.append(record)

1008

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

1009

1010

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1011

("records",

1012

"output"))

1013

1014

1015

class Unique(object):

1016

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

1017

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

1018

1019

1020

class Test_string_to_delta(TestCaseWithAssertLogs):

1021

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

1022

# rfc3339_duration_to_delta() already has more comprehensive

1023

# tests, which are run by doctest.

1024

1025

def test_rfc3339_zero_seconds(self):

1026

self.assertEqual(datetime.timedelta(),

1027

string_to_delta("PT0S"))

1028

1029

def test_rfc3339_zero_days(self):

1030

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

1031

1032

def test_rfc3339_one_second(self):

1033

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

1034

string_to_delta("PT1S"))

1035

1036

def test_rfc3339_two_hours(self):

1037

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

1038

string_to_delta("PT2H"))

1039

1040

def test_falls_back_to_pre_1_6_1_with_warning(self):

1041

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

1042

value = string_to_delta("2h")

1043

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

1044

1045

1046

class Test_check_option_syntax(unittest.TestCase):

1047

def setUp(self):

1048

self.parser = argparse.ArgumentParser()

1049

add_command_line_options(self.parser)

1050

1051

def test_actions_requires_client_or_all(self):

1052

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

1053

args = self.actionargs(action, value)

1054

with self.assertParseError():

1055

self.parse_args(args)

1056

1057

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

1058

# check_option_syntax()

1059

actions = {

1060

"--enable": None,

1061

"--disable": None,

1062

"--bump-timeout": None,

1063

"--start-checker": None,

1064

"--stop-checker": None,

1065

"--is-enabled": None,

1066

"--remove": None,

1067

"--checker": "x",

1068

"--timeout": "PT0S",

1069

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

1070

"--interval": "PT0S",

1071

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

1072

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

1073

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

1074

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

1075

"--host": "hostname",

1076

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

1077

"--approve": None,

1078

"--deny": None,

1079

}

1080

1081

@staticmethod

1082

def actionargs(action, value, *args):

1083

if value is not None:

1084

return [action, value] + list(args)

1085

else:

1086

return [action] + list(args)

1087

1088

@contextlib.contextmanager

1089

def assertParseError(self):

1090

with self.assertRaises(SystemExit) as e:

1091

with self.redirect_stderr_to_devnull():

1092

yield

1093

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

1094

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

1095

# /argparse.html#exiting-methods

1096

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

1097

1098

def parse_args(self, args):

1099

options = self.parser.parse_args(args)

1100

check_option_syntax(self.parser, options)

1101

1102

@staticmethod

1103

@contextlib.contextmanager

1104

def redirect_stderr_to_devnull():

1105

old_stderr = sys.stderr

1106

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

1107

sys.stderr = null

1108

try:

1109

yield

1110

finally:

1111

sys.stderr = old_stderr

1112

1113

def check_option_syntax(self, options):

1114

check_option_syntax(self.parser, options)

1115

1116

def test_actions_all_conflicts_with_verbose(self):

1117

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

1118

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

1119

"--verbose")

1120

with self.assertParseError():

1121

self.parse_args(args)

1122

1123

def test_actions_with_client_conflicts_with_verbose(self):

1124

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

1125

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

1126

"client")

1127

with self.assertParseError():

1128

self.parse_args(args)

1129

1130

def test_dump_json_conflicts_with_verbose(self):

1131

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

1132

with self.assertParseError():

1133

self.parse_args(args)

1134

1135

def test_dump_json_conflicts_with_action(self):

1136

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

1137

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

1138

with self.assertParseError():

1139

self.parse_args(args)

1140

1141

def test_all_can_not_be_alone(self):

1142

args = ["--all"]

1143

with self.assertParseError():

1144

self.parse_args(args)

1145

1146

def test_all_is_ok_with_any_action(self):

1147

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

1148

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

1149

self.parse_args(args)

1150

1151

def test_any_action_is_ok_with_one_client(self):

1152

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

1153

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

1154

self.parse_args(args)

1155

1156

def test_one_client_with_all_actions_except_is_enabled(self):

1157

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

1158

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

1159

continue

1160

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

1161

self.parse_args(args)

1162

1163

def test_two_clients_with_all_actions_except_is_enabled(self):

1164

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

1165

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

1166

continue

1167

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

1168

"client2")

1169

self.parse_args(args)

1170

1171

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1172

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

1173

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

1174

continue

1175

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

1176

"client2")

1177

self.parse_args(args)

1178

1179

def test_is_enabled_fails_without_client(self):

1180

args = ["--is-enabled"]

1181

with self.assertParseError():

1182

self.parse_args(args)

1183

1184

def test_is_enabled_fails_with_two_clients(self):

1185

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

1186

with self.assertParseError():

1187

self.parse_args(args)

1188

1189

def test_remove_can_only_be_combined_with_action_deny(self):

1190

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

1191

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

1192

continue

1193

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

1194

"--remove")

1195

with self.assertParseError():

1196

self.parse_args(args)

1197

1198

1199

class Test_dbus_exceptions(unittest.TestCase):

1200

1201

def test_dbus_ConnectFailed_is_Error(self):

1202

with self.assertRaises(dbus.Error):

1203

raise dbus.ConnectFailed()

1204

1205

1206

class Test_dbus_MandosBus(unittest.TestCase):

1207

1208

class MockMandosBus(dbus.MandosBus):

1209

def __init__(self):

1210

self._name = "se.recompile.Mandos"

1211

self._server_path = "/"

1212

self._server_interface = "se.recompile.Mandos"

1213

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

1214

self.calls = []

1215

self.call_method_return = Unique()

1216

1217

def call_method(self, methodname, busname, objectpath,

1218

interface, *args):

1219

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

1220

interface, args))

1221

return self.call_method_return

1222

1223

def setUp(self):

1224

self.bus = self.MockMandosBus()

1225

1226

def test_set_client_property(self):

1227

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

1228

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

1229

"org.freedesktop.DBus.Properties",

1230

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

1231

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

1232

1233

def test_call_client_method(self):

1234

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

1235

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

1236

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

1237

self.bus._client_interface, ())

1238

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

1239

1240

def test_call_client_method_with_args(self):

1241

args = (Unique(), Unique())

1242

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

1243

*args)

1244

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

1245

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

1246

self.bus._client_interface,

1247

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

1248

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

1249

1250

def test_get_clients_and_properties(self):

1251

managed_objects = {

1252

"objectpath": {

1253

self.bus._client_interface: {

1254

"key": "value",

1255

"bool": True,

1256

},

1257

"irrelevant_interface": {

1258

"key": "othervalue",

1259

"bool": False,

1260

},

1261

},

1262

"other_objectpath": {

1263

"other_irrelevant_interface": {

1264

"key": "value 3",

1265

"bool": None,

1266

},

1267

},

1268

}

1269

expected_clients_and_properties = {

1270

"objectpath": {

1271

"key": "value",

1272

"bool": True,

1273

}

1274

}

1275

self.bus.call_method_return = managed_objects

1276

ret = self.bus.get_clients_and_properties()

1277

self.assertDictEqual(expected_clients_and_properties, ret)

1278

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

1279

self.bus._server_path,

1280

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

1281

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

1282

1283

def test_call_server_method(self):

1284

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

1285

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

1286

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

1287

self.bus._server_path,

1288

self.bus._server_interface, ())

1289

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

1290

1291

def test_call_server_method_with_args(self):

1292

args = (Unique(), Unique())

1293

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

1294

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

1295

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

1296

self.bus._server_path,

1297

self.bus._server_interface,

1298

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

1299

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

1300

1301

1302

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1303

1304

def MockDBusPython_func(self, func):

1305

class mock_dbus_python(object):

1306

"""mock dbus-python module"""

1307

class exceptions(object):

1308

"""Pseudo-namespace"""

1309

class DBusException(Exception):

1310

pass

1311

class SystemBus(object):

1312

@staticmethod

1313

def get_object(busname, objectpath):

1314

DBusObject = collections.namedtuple(

1315

"DBusObject", ("methodname",))

1316

def method(*args, **kwargs):

1317

self.assertEqual({"dbus_interface":

1318

"interface"},

1319

kwargs)

1320

return func(*args)

1321

return DBusObject(methodname=method)

1322

class Boolean(object):

1323

def __init__(self, value):

1324

self.value = bool(value)

1325

def __bool__(self):

1326

return self.value

1327

if sys.version_info.major == 2:

1328

__nonzero__ = __bool__

1329

class ObjectPath(str):

1330

pass

1331

class Dictionary(dict):

1332

pass

1333

return mock_dbus_python

1334

1335

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

1336

interface, *args):

1337

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

1338

return bus.call_method(methodname, busname, objectpath,

1339

interface, *args)

1340

1341

def test_call_method_returns(self):

1342

expected_method_return = Unique()

1343

method_args = (Unique(), Unique())

1344

def func(*args):

1345

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

1346

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

1347

self.assertIs(marg, arg)

1348

return expected_method_return

1349

mock_dbus_python = self.MockDBusPython_func(func)

1350

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1351

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

1352

"objectpath", "interface",

1353

*method_args)

1354

self.assertIs(ret, expected_method_return)

1355

1356

def test_call_method_filters_bool_true(self):

1357

def func():

1358

return method_return

1359

mock_dbus_python = self.MockDBusPython_func(func)

1360

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1361

method_return = mock_dbus_python.Boolean(True)

1362

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

1363

"objectpath", "interface")

1364

self.assertTrue(ret)

1365

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1366

1367

def test_call_method_filters_bool_false(self):

1368

def func():

1369

return method_return

1370

mock_dbus_python = self.MockDBusPython_func(func)

1371

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1372

method_return = mock_dbus_python.Boolean(False)

1373

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

1374

"objectpath", "interface")

1375

self.assertFalse(ret)

1376

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1377

1378

def test_call_method_filters_objectpath(self):

1379

def func():

1380

return method_return

1381

mock_dbus_python = self.MockDBusPython_func(func)

1382

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1383

method_return = mock_dbus_python.ObjectPath("objectpath")

1384

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

1385

"objectpath", "interface")

1386

self.assertEqual("objectpath", ret)

1387

self.assertIsNot("objectpath", ret)

1388

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1389

1390

def test_call_method_filters_booleans_in_dict(self):

1391

def func():

1392

return method_return

1393

mock_dbus_python = self.MockDBusPython_func(func)

1394

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1395

method_return = mock_dbus_python.Dictionary(

1396

{mock_dbus_python.Boolean(True):

1397

mock_dbus_python.Boolean(False),

1398

mock_dbus_python.Boolean(False):

1399

mock_dbus_python.Boolean(True)})

1400

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

1401

"objectpath", "interface")

1402

expected_method_return = {True: False,

1403

False: True}

1404

self.assertEqual(expected_method_return, ret)

1405

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1406

1407

def test_call_method_filters_objectpaths_in_dict(self):

1408

def func():

1409

return method_return

1410

mock_dbus_python = self.MockDBusPython_func(func)

1411

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1412

method_return = mock_dbus_python.Dictionary(

1413

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1414

mock_dbus_python.ObjectPath("objectpath_value_1"),

1415

mock_dbus_python.ObjectPath("objectpath_key_2"):

1416

mock_dbus_python.ObjectPath("objectpath_value_2")})

1417

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

1418

"objectpath", "interface")

1419

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

1420

for key, value in

1421

method_return.items()}

1422

self.assertEqual(expected_method_return, ret)

1423

self.assertIsInstance(ret, dict)

1424

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1425

1426

def test_call_method_filters_dict_in_dict(self):

1427

def func():

1428

return method_return

1429

mock_dbus_python = self.MockDBusPython_func(func)

1430

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1431

method_return = mock_dbus_python.Dictionary(

1432

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

1433

"key12": "value12"}),

1434

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

1435

"key22": "value22"})})

1436

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

1437

"objectpath", "interface")

1438

expected_method_return = {

1439

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

1440

"key12": "value12"},

1441

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

1442

"key22": "value22"},

1443

}

1444

self.assertEqual(expected_method_return, ret)

1445

self.assertIsInstance(ret, dict)

1446

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1447

for key, value in ret.items():

1448

self.assertIsInstance(value, dict)

1449

self.assertEqual(expected_method_return[key], value)

1450

self.assertNotIsInstance(value,

1451

mock_dbus_python.Dictionary)

1452

1453

def test_call_method_filters_dict_three_deep(self):

1454

def func():

1455

return method_return

1456

mock_dbus_python = self.MockDBusPython_func(func)

1457

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1458

method_return = mock_dbus_python.Dictionary(

1459

{"key1":

1460

mock_dbus_python.Dictionary(

1461

{"key2":

1462

mock_dbus_python.Dictionary(

1463

{"key3":

1464

mock_dbus_python.Boolean(True),

1465

}),

1466

}),

1467

})

1468

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

1469

"objectpath", "interface")

1470

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

1471

self.assertEqual(expected_method_return, ret)

1472

self.assertIsInstance(ret, dict)

1473

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1474

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

1475

self.assertNotIsInstance(ret["key1"],

1476

mock_dbus_python.Dictionary)

1477

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

1478

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

1479

mock_dbus_python.Dictionary)

1480

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

1481

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

1482

mock_dbus_python.Boolean)

1483

1484

def test_call_method_handles_exception(self):

1485

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

1486

1487

def func():

1488

dbus_logger.error("Test")

1489

raise mock_dbus_python.exceptions.DBusException()

1490

1491

mock_dbus_python = self.MockDBusPython_func(func)

1492

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1493

1494

class CountingHandler(logging.Handler):

1495

count = 0

1496

def emit(self, record):

1497

self.count += 1

1498

1499

counting_handler = CountingHandler()

1500

1501

dbus_logger.addHandler(counting_handler)

1502

1503

try:

1504

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

1505

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

1506

"objectpath", "interface")

1507

finally:

1508

dbus_logger.removeFilter(counting_handler)

1509

1510

self.assertNotIsInstance(e, dbus.ConnectFailed)

1511

1512

# Make sure the dbus logger was suppressed

1513

self.assertEqual(0, counting_handler.count)

1514

1515

def test_get_object_converts_to_correct_exception(self):

1516

bus = dbus_python_adapter.SystemBus(

1517

self.fake_dbus_python_raises_exception_on_connect)

1518

with self.assertRaises(dbus.ConnectFailed):

1519

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

1520

"objectpath", "interface")

1521

1522

class fake_dbus_python_raises_exception_on_connect(object):

1523

"""fake dbus-python module"""

1524

class exceptions(object):

1525

"""Pseudo-namespace"""

1526

class DBusException(Exception):

1527

pass

1528

1529

@classmethod

1530

def SystemBus(cls):

1531

def get_object(busname, objectpath):

1532

raise cls.exceptions.DBusException()

1533

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

1534

return Bus(get_object=get_object)

1535

1536

1537

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1538

class mock_dbus_python(object):

1539

"""mock dbus-python modules"""

1540

class SystemBus(object):

1541

@staticmethod

1542

def get_object(busname, objectpath):

1543

return Unique()

1544

1545

def setUp(self):

1546

self.bus = dbus_python_adapter.CachingBus(

1547

self.mock_dbus_python)

1548

1549

def test_returns_distinct_objectpaths(self):

1550

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

1551

self.assertIsInstance(obj1, Unique)

1552

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

1553

self.assertIsInstance(obj2, Unique)

1554

self.assertIsNot(obj1, obj2)

1555

1556

def test_returns_distinct_busnames(self):

1557

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

1558

self.assertIsInstance(obj1, Unique)

1559

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

1560

self.assertIsInstance(obj2, Unique)

1561

self.assertIsNot(obj1, obj2)

1562

1563

def test_returns_distinct_both(self):

1564

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

1565

self.assertIsInstance(obj1, Unique)

1566

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

1567

self.assertIsInstance(obj2, Unique)

1568

self.assertIsNot(obj1, obj2)

1569

1570

def test_returns_same(self):

1571

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

1572

self.assertIsInstance(obj1, Unique)

1573

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

1574

self.assertIsInstance(obj2, Unique)

1575

self.assertIs(obj1, obj2)

1576

1577

def test_returns_same_old(self):

1578

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

1579

self.assertIsInstance(obj1, Unique)

1580

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

1581

self.assertIsInstance(obj2, Unique)

1582

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

1583

self.assertIsInstance(obj1b, Unique)

1584

self.assertIsNot(obj1, obj2)

1585

self.assertIsNot(obj2, obj1b)

1586

self.assertIs(obj1, obj1b)

1587

1588

1589

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1590

1591

def Stub_pydbus_func(self, func):

1592

class stub_pydbus(object):

1593

"""stub pydbus module"""

1594

class SystemBus(object):

1595

@staticmethod

1596

def get(busname, objectpath):

1597

DBusObject = collections.namedtuple(

1598

"DBusObject", ("methodname",))

1599

return {"interface":

1600

DBusObject(methodname=func)}

1601

return stub_pydbus

1602

1603

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

1604

interface, *args):

1605

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

1606

return bus.call_method(methodname, busname, objectpath,

1607

interface, *args)

1608

1609

def test_call_method_returns(self):

1610

expected_method_return = Unique()

1611

method_args = (Unique(), Unique())

1612

def func(*args):

1613

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

1614

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

1615

self.assertIs(marg, arg)

1616

return expected_method_return

1617

stub_pydbus = self.Stub_pydbus_func(func)

1618

bus = pydbus_adapter.SystemBus(stub_pydbus)

1619

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

1620

"objectpath", "interface",

1621

*method_args)

1622

self.assertIs(ret, expected_method_return)

1623

1624

def test_call_method_handles_exception(self):

1625

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

1626

1627

def func():

1628

raise gi.repository.GLib.Error()

1629

1630

stub_pydbus = self.Stub_pydbus_func(func)

1631

bus = pydbus_adapter.SystemBus(stub_pydbus)

1632

1633

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

1634

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

1635

"objectpath", "interface")

1636

1637

self.assertNotIsInstance(e, dbus.ConnectFailed)

1638

1639

def test_get_converts_to_correct_exception(self):

1640

bus = pydbus_adapter.SystemBus(

1641

self.fake_pydbus_raises_exception_on_connect)

1642

with self.assertRaises(dbus.ConnectFailed):

1643

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

1644

"objectpath", "interface")

1645

1646

class fake_pydbus_raises_exception_on_connect(object):

1647

"""fake dbus-python module"""

1648

@classmethod

1649

def SystemBus(cls):

1650

def get(busname, objectpath):

1651

raise gi.repository.GLib.Error()

1652

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

1653

return Bus(get=get)

1654

1655

def test_set_property_uses_setattr(self):

1656

class Object(object):

1657

pass

1658

obj = Object()

1659

class pydbus_spy(object):

1660

class SystemBus(object):

1661

@staticmethod

1662

def get(busname, objectpath):

1663

return {"interface": obj}

1664

bus = pydbus_adapter.SystemBus(pydbus_spy)

1665

value = Unique()

1666

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

1667

value)

1668

self.assertIs(value, obj.key)

1669

1670

def test_get_suppresses_xml_deprecation_warning(self):

1671

if sys.version_info.major >= 3:

1672

return

1673

class stub_pydbus_get(object):

1674

class SystemBus(object):

1675

@staticmethod

1676

def get(busname, objectpath):

1677

warnings.warn_explicit(

1678

"deprecated", DeprecationWarning,

1679

"xml.etree.ElementTree", 0)

1680

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1681

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

1682

warnings.simplefilter("always")

1683

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

1684

self.assertEqual(0, len(w))

1685

1686

1687

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1688

class stub_pydbus(object):

1689

"""stub pydbus module"""

1690

class SystemBus(object):

1691

@staticmethod

1692

def get(busname, objectpath):

1693

return Unique()

1694

1695

def setUp(self):

1696

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

1697

1698

def test_returns_distinct_objectpaths(self):

1699

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

1700

self.assertIsInstance(obj1, Unique)

1701

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

1702

self.assertIsInstance(obj2, Unique)

1703

self.assertIsNot(obj1, obj2)

1704

1705

def test_returns_distinct_busnames(self):

1706

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

1707

self.assertIsInstance(obj1, Unique)

1708

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

1709

self.assertIsInstance(obj2, Unique)

1710

self.assertIsNot(obj1, obj2)

1711

1712

def test_returns_distinct_both(self):

1713

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

1714

self.assertIsInstance(obj1, Unique)

1715

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

1716

self.assertIsInstance(obj2, Unique)

1717

self.assertIsNot(obj1, obj2)

1718

1719

def test_returns_same(self):

1720

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

1721

self.assertIsInstance(obj1, Unique)

1722

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

1723

self.assertIsInstance(obj2, Unique)

1724

self.assertIs(obj1, obj2)

1725

1726

def test_returns_same_old(self):

1727

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

1728

self.assertIsInstance(obj1, Unique)

1729

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

1730

self.assertIsInstance(obj2, Unique)

1731

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

1732

self.assertIsInstance(obj1b, Unique)

1733

self.assertIsNot(obj1, obj2)

1734

self.assertIsNot(obj2, obj1b)

1735

self.assertIs(obj1, obj1b)

1736

1737

1738

class Test_commands_from_options(unittest.TestCase):

1739

1740

def setUp(self):

1741

self.parser = argparse.ArgumentParser()

1742

add_command_line_options(self.parser)

1743

1744

def test_is_enabled(self):

1745

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

1746

command.IsEnabled)

1747

1748

def assert_command_from_args(self, args, command_cls,

1749

**cmd_attrs):

1750

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

1751

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

1752

options = self.parser.parse_args(args)

1753

check_option_syntax(self.parser, options)

1754

commands = commands_from_options(options)

1755

self.assertEqual(1, len(commands))

1756

command = commands[0]

1757

self.assertIsInstance(command, command_cls)

1758

for key, value in cmd_attrs.items():

1759

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

1760

1761

def test_is_enabled_short(self):

1762

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

1763

command.IsEnabled)

1764

1765

def test_approve(self):

1766

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

1767

command.Approve)

1768

1769

def test_approve_short(self):

1770

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

1771

command.Approve)

1772

1773

def test_deny(self):

1774

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

1775

command.Deny)

1776

1777

def test_deny_short(self):

1778

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

1779

1780

def test_remove(self):

1781

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

1782

command.Remove)

1783

1784

def test_deny_before_remove(self):

1785

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

1786

"client"])

1787

check_option_syntax(self.parser, options)

1788

commands = commands_from_options(options)

1789

self.assertEqual(2, len(commands))

1790

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

1791

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

1792

1793

def test_deny_before_remove_reversed(self):

1794

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

1795

"--all"])

1796

check_option_syntax(self.parser, options)

1797

commands = commands_from_options(options)

1798

self.assertEqual(2, len(commands))

1799

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

1800

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

1801

1802

def test_remove_short(self):

1803

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

1804

command.Remove)

1805

1806

def test_dump_json(self):

1807

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

1808

command.DumpJSON)

1809

1810

def test_enable(self):

1811

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

1812

command.Enable)

1813

1814

def test_enable_short(self):

1815

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

1816

command.Enable)

1817

1818

def test_disable(self):

1819

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

1820

command.Disable)

1821

1822

def test_disable_short(self):

1823

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

1824

command.Disable)

1825

1826

def test_bump_timeout(self):

1827

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

1828

command.BumpTimeout)

1829

1830

def test_bump_timeout_short(self):

1831

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

1832

command.BumpTimeout)

1833

1834

def test_start_checker(self):

1835

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

1836

command.StartChecker)

1837

1838

def test_stop_checker(self):

1839

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

1840

command.StopChecker)

1841

1842

def test_approve_by_default(self):

1843

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

1844

"client"],

1845

command.ApproveByDefault)

1846

1847

def test_deny_by_default(self):

1848

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

1849

command.DenyByDefault)

1850

1851

def test_checker(self):

1852

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

1853

command.SetChecker,

1854

value_to_set=":")

1855

1856

def test_checker_empty(self):

1857

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

1858

command.SetChecker,

1859

value_to_set="")

1860

1861

def test_checker_short(self):

1862

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

1863

command.SetChecker,

1864

value_to_set=":")

1865

1866

def test_host(self):

1867

self.assert_command_from_args(

1868

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

1869

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

1870

1871

def test_host_short(self):

1872

self.assert_command_from_args(

1873

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

1874

value_to_set="client.example.org")

1875

1876

def test_secret_devnull(self):

1877

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

1878

"client"], command.SetSecret,

1879

value_to_set=b"")

1880

1881

def test_secret_tempfile(self):

1882

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

1883

value = b"secret\0xyzzy\nbar"

1884

f.write(value)

1885

f.seek(0)

1886

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

1887

"client"],

1888

command.SetSecret,

1889

value_to_set=value)

1890

1891

def test_secret_devnull_short(self):

1892

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

1893

"client"], command.SetSecret,

1894

value_to_set=b"")

1895

1896

def test_secret_tempfile_short(self):

1897

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

1898

value = b"secret\0xyzzy\nbar"

1899

f.write(value)

1900

f.seek(0)

1901

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

1902

command.SetSecret,

1903

value_to_set=value)

1904

1905

def test_timeout(self):

1906

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

1907

command.SetTimeout,

1908

value_to_set=300000)

1909

1910

def test_timeout_short(self):

1911

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

1912

command.SetTimeout,

1913

value_to_set=300000)

1914

1915

def test_extended_timeout(self):

1916

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

1917

"client"],

1918

command.SetExtendedTimeout,

1919

value_to_set=900000)

1920

1921

def test_interval(self):

1922

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

1923

"client"], command.SetInterval,

1924

value_to_set=120000)

1925

1926

def test_interval_short(self):

1927

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

1928

command.SetInterval,

1929

value_to_set=120000)

1930

1931

def test_approval_delay(self):

1932

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

1933

"client"],

1934

command.SetApprovalDelay,

1935

value_to_set=30000)

1936

1937

def test_approval_duration(self):

1938

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

1939

"client"],

1940

command.SetApprovalDuration,

1941

value_to_set=1000)

1942

1943

def test_print_table(self):

1944

self.assert_command_from_args([], command.PrintTable,

1945

verbose=False)

1946

1947

def test_print_table_verbose(self):

1948

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

1949

command.PrintTable,

1950

verbose=True)

1951

1952

def test_print_table_verbose_short(self):

1953

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

1954

verbose=True)

1955

1956

1957

class TestCommand(unittest.TestCase):

1958

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

1959

1960

class FakeMandosBus(dbus.MandosBus):

1961

def __init__(self, testcase):

1962

self.client_properties = {

1963

"Name": "foo",

1964

"KeyID": ("92ed150794387c03ce684574b1139a65"

1965

"94a34f895daaaf09fd8ea90a27cddb12"),

1966

"Secret": b"secret",

1967

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

1968

"Enabled": True,

1969

"Timeout": 300000,

1970

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

1971

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

1972

"Interval": 120000,

1973

"Fingerprint": ("778827225BA7DE539C5A"

1974

"7CFA59CFF7CDBD9A5920"),

1975

"CheckerRunning": False,

1976

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

1977

"ApprovalPending": False,

1978

"ApprovedByDefault": True,

1979

"LastApprovalRequest": "",

1980

"ApprovalDelay": 0,

1981

"ApprovalDuration": 1000,

1982

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

1983

"ExtendedTimeout": 900000,

1984

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

1985

"LastCheckerStatus": 0,

1986

}

1987

self.other_client_properties = {

1988

"Name": "barbar",

1989

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1990

"6ab612cff5ad227247e46c2b020f441c"),

1991

"Secret": b"secretbar",

1992

"Host": "192.0.2.3",

1993

"Enabled": True,

1994

"Timeout": 300000,

1995

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

1996

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

1997

"Interval": 120000,

1998

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1999

"F547B3A107558FCA3A27"),

2000

"CheckerRunning": True,

2001

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

2002

"ApprovalPending": False,

2003

"ApprovedByDefault": False,

2004

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

2005

"ApprovalDelay": 30000,

2006

"ApprovalDuration": 93785000,

2007

"Checker": ":",

2008

"ExtendedTimeout": 900000,

2009

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

2010

"LastCheckerStatus": -2,

2011

}

2012

self.clients = collections.OrderedDict(

2013

[

2014

("client_objectpath", self.client_properties),

2015

("other_client_objectpath",

2016

self.other_client_properties),

2017

])

2018

self.one_client = {"client_objectpath":

2019

self.client_properties}

2020

self.testcase = testcase

2021

self.calls = []

2022

2023

def call_method(self, methodname, busname, objectpath,

2024

interface, *args):

2025

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

2026

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

2027

interface, args))

2028

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

2029

if methodname == "Set":

2030

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

2031

interface, key, value = args

2032

self.testcase.assertEqual(

2033

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

2034

self.clients[objectpath][key] = value

2035

return

2036

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

2037

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

2038

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

2039

clientpath = args[0]

2040

del self.clients[clientpath]

2041

return

2042

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

2043

if methodname == "Approve":

2044

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

2045

return

2046

raise ValueError()

2047

2048

def setUp(self):

2049

self.bus = self.FakeMandosBus(self)

2050

2051

2052

class TestBaseCommands(TestCommand):

2053

2054

def test_IsEnabled_exits_successfully(self):

2055

with self.assertRaises(SystemExit) as e:

2056

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

2057

if e.exception.code is not None:

2058

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

2059

else:

2060

self.assertIsNone(e.exception.code)

2061

2062

def test_IsEnabled_exits_with_failure(self):

2063

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

2064

with self.assertRaises(SystemExit) as e:

2065

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

2066

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

2067

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

2068

else:

2069

self.assertIsNotNone(e.exception.code)

2070

2071

def test_Approve(self):

2072

busname = "se.recompile.Mandos"

2073

client_interface = "se.recompile.Mandos.Client"

2074

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

2075

for clientpath in self.bus.clients:

2076

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

2077

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

2078

2079

def test_Deny(self):

2080

busname = "se.recompile.Mandos"

2081

client_interface = "se.recompile.Mandos.Client"

2082

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

2083

for clientpath in self.bus.clients:

2084

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

2085

client_interface, (False,)),

2086

self.bus.calls)

2087

2088

def test_Remove(self):

2089

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

2090

for clientpath in self.bus.clients:

2091

self.assertIn(("RemoveClient", dbus_busname,

2092

dbus_server_path, dbus_server_interface,

2093

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

2094

2095

expected_json = {

2096

"foo": {

2097

"Name": "foo",

2098

"KeyID": ("92ed150794387c03ce684574b1139a65"

2099

"94a34f895daaaf09fd8ea90a27cddb12"),

2100

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

2101

"Enabled": True,

2102

"Timeout": 300000,

2103

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

2104

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

2105

"Interval": 120000,

2106

"Fingerprint": ("778827225BA7DE539C5A"

2107

"7CFA59CFF7CDBD9A5920"),

2108

"CheckerRunning": False,

2109

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

2110

"ApprovalPending": False,

2111

"ApprovedByDefault": True,

2112

"LastApprovalRequest": "",

2113

"ApprovalDelay": 0,

2114

"ApprovalDuration": 1000,

2115

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

2116

"ExtendedTimeout": 900000,

2117

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

2118

"LastCheckerStatus": 0,

2119

},

2120

"barbar": {

2121

"Name": "barbar",

2122

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2123

"6ab612cff5ad227247e46c2b020f441c"),

2124

"Host": "192.0.2.3",

2125

"Enabled": True,

2126

"Timeout": 300000,

2127

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

2128

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

2129

"Interval": 120000,

2130

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2131

"F547B3A107558FCA3A27"),

2132

"CheckerRunning": True,

2133

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

2134

"ApprovalPending": False,

2135

"ApprovedByDefault": False,

2136

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

2137

"ApprovalDelay": 30000,

2138

"ApprovalDuration": 93785000,

2139

"Checker": ":",

2140

"ExtendedTimeout": 900000,

2141

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

2142

"LastCheckerStatus": -2,

2143

},

2144

}

2145

2146

def test_DumpJSON_normal(self):

2147

with self.capture_stdout_to_buffer() as buffer:

2148

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

2149

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

2150

self.assertDictEqual(self.expected_json, json_data)

2151

2152

@staticmethod

2153

@contextlib.contextmanager

2154

def capture_stdout_to_buffer():

2155

capture_buffer = io.StringIO()

2156

old_stdout = sys.stdout

2157

sys.stdout = capture_buffer

2158

try:

2159

yield capture_buffer

2160

finally:

2161

sys.stdout = old_stdout

2162

2163

def test_DumpJSON_one_client(self):

2164

with self.capture_stdout_to_buffer() as buffer:

2165

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

2166

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

2167

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

2168

self.assertDictEqual(expected_json, json_data)

2169

2170

def test_PrintTable_normal(self):

2171

with self.capture_stdout_to_buffer() as buffer:

2172

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

2173

expected_output = "\n".join((

2174

"Name Enabled Timeout Last Successful Check",

2175

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

2176

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

2177

)) + "\n"

2178

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

2179

2180

def test_PrintTable_verbose(self):

2181

with self.capture_stdout_to_buffer() as buffer:

2182

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

2183

columns = (

2184

(

2185

"Name ",

2186

"foo ",

2187

"barbar ",

2188

),(

2189

"Enabled ",

2190

"Yes ",

2191

"Yes ",

2192

),(

2193

"Timeout ",

2194

"00:05:00 ",

2195

"00:05:00 ",

2196

),(

2197

"Last Successful Check ",

2198

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

2199

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

2200

),(

2201

"Created ",

2202

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

2203

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

2204

),(

2205

"Interval ",

2206

"00:02:00 ",

2207

"00:02:00 ",

2208

),(

2209

"Host ",

2210

"foo.example.org ",

2211

"192.0.2.3 ",

2212

),(

2213

("Key ID "

2214

" "),

2215

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2216

"ea90a27cddb12 "),

2217

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2218

"46c2b020f441c "),

2219

),(

2220

"Fingerprint ",

2221

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2222

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2223

),(

2224

"Check Is Running ",

2225

"No ",

2226

"Yes ",

2227

),(

2228

"Last Enabled ",

2229

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

2230

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

2231

),(

2232

"Approval Is Pending ",

2233

"No ",

2234

"No ",

2235

),(

2236

"Approved By Default ",

2237

"Yes ",

2238

"No ",

2239

),(

2240

"Last Approval Request ",

2241

" ",

2242

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

2243

),(

2244

"Approval Delay ",

2245

"00:00:00 ",

2246

"00:00:30 ",

2247

),(

2248

"Approval Duration ",

2249

"00:00:01 ",

2250

"1T02:03:05 ",

2251

),(

2252

"Checker ",

2253

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

2254

": ",

2255

),(

2256

"Extended Timeout ",

2257

"00:15:00 ",

2258

"00:15:00 ",

2259

),(

2260

"Expires ",

2261

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

2262

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

2263

),(

2264

"Last Checker Status",

2265

"0 ",

2266

"-2 ",

2267

)

2268

)

2269

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

2270

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

2271

for rows in columns)

2272

for line in range(num_lines))

2273

+ "\n")

2274

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

2275

2276

def test_PrintTable_one_client(self):

2277

with self.capture_stdout_to_buffer() as buffer:

2278

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

2279

expected_output = "\n".join((

2280

"Name Enabled Timeout Last Successful Check",

2281

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

2282

)) + "\n"

2283

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

2284

2285

2286

class TestPropertySetterCmd(TestCommand):

2287

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

2288

2289

def runTest(self):

2290

if not hasattr(self, "command"):

2291

return # Abstract TestCase class

2292

2293

if hasattr(self, "values_to_set"):

2294

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

2295

values_to_get = getattr(self, "values_to_get",

2296

self.values_to_set)

2297

else:

2298

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

2299

values_to_get = self.values_to_get

2300

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

2301

for clientpath in self.bus.clients:

2302

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

2303

Unique())

2304

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

2305

for clientpath in self.bus.clients:

2306

value = (self.bus.clients[clientpath]

2307

[self.propname])

2308

self.assertNotIsInstance(value, Unique)

2309

self.assertEqual(value_to_get, value)

2310

2311

2312

class TestEnableCmd(TestPropertySetterCmd):

2313

command = command.Enable

2314

propname = "Enabled"

2315

values_to_get = [True]

2316

2317

2318

class TestDisableCmd(TestPropertySetterCmd):

2319

command = command.Disable

2320

propname = "Enabled"

2321

values_to_get = [False]

2322

2323

2324

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2325

command = command.BumpTimeout

2326

propname = "LastCheckedOK"

2327

values_to_get = [""]

2328

2329

2330

class TestStartCheckerCmd(TestPropertySetterCmd):

2331

command = command.StartChecker

2332

propname = "CheckerRunning"

2333

values_to_get = [True]

2334

2335

2336

class TestStopCheckerCmd(TestPropertySetterCmd):

2337

command = command.StopChecker

2338

propname = "CheckerRunning"

2339

values_to_get = [False]

2340

2341

2342

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2343

command = command.ApproveByDefault

2344

propname = "ApprovedByDefault"

2345

values_to_get = [True]

2346

2347

2348

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2349

command = command.DenyByDefault

2350

propname = "ApprovedByDefault"

2351

values_to_get = [False]

2352

2353

2354

class TestSetCheckerCmd(TestPropertySetterCmd):

2355

command = command.SetChecker

2356

propname = "Checker"

2357

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

2358

2359

2360

class TestSetHostCmd(TestPropertySetterCmd):

2361

command = command.SetHost

2362

propname = "Host"

2363

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

2364

2365

2366

class TestSetSecretCmd(TestPropertySetterCmd):

2367

command = command.SetSecret

2368

propname = "Secret"

2369

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

2370

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

2371

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

2372

2373

2374

class TestSetTimeoutCmd(TestPropertySetterCmd):

2375

command = command.SetTimeout

2376

propname = "Timeout"

2377

values_to_set = [datetime.timedelta(),

2378

datetime.timedelta(minutes=5),

2379

datetime.timedelta(seconds=1),

2380

datetime.timedelta(weeks=1),

2381

datetime.timedelta(weeks=52)]

2382

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

2383

2384

2385

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2386

command = command.SetExtendedTimeout

2387

propname = "ExtendedTimeout"

2388

values_to_set = [datetime.timedelta(),

2389

datetime.timedelta(minutes=5),

2390

datetime.timedelta(seconds=1),

2391

datetime.timedelta(weeks=1),

2392

datetime.timedelta(weeks=52)]

2393

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

2394

2395

2396

class TestSetIntervalCmd(TestPropertySetterCmd):

2397

command = command.SetInterval

2398

propname = "Interval"

2399

values_to_set = [datetime.timedelta(),

2400

datetime.timedelta(minutes=5),

2401

datetime.timedelta(seconds=1),

2402

datetime.timedelta(weeks=1),

2403

datetime.timedelta(weeks=52)]

2404

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

2405

2406

2407

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2408

command = command.SetApprovalDelay

2409

propname = "ApprovalDelay"

2410

values_to_set = [datetime.timedelta(),

2411

datetime.timedelta(minutes=5),

2412

datetime.timedelta(seconds=1),

2413

datetime.timedelta(weeks=1),

2414

datetime.timedelta(weeks=52)]

2415

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

2416

2417

2418

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2419

command = command.SetApprovalDuration

2420

propname = "ApprovalDuration"

2421

values_to_set = [datetime.timedelta(),

2422

datetime.timedelta(minutes=5),

2423

datetime.timedelta(seconds=1),

2424

datetime.timedelta(weeks=1),

2425

datetime.timedelta(weeks=52)]

2426

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

2427

2428

2429

2430

def should_only_run_tests():

2431

parser = argparse.ArgumentParser(add_help=False)

2432

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

2433

args, unknown_args = parser.parse_known_args()

2434

run_tests = args.check

2435

if run_tests:

2436

# Remove --check argument from sys.argv

2437

sys.argv[1:] = unknown_args

2438

return run_tests

2439

2440

# Add all tests from doctest strings

2441

def load_tests(loader, tests, none):

2442

import doctest

2443

tests.addTests(doctest.DocTestSuite())

2444

return tests

2445

2446

if __name__ == "__main__":

2447

try:

2448

if should_only_run_tests():

2449

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

2450

unittest.main()

2451

else:

2452

main()

2453

finally:

2454

logging.shutdown()