/mandos/trunk : revision 1065

1

#!/usr/bin/python

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

37

import locale

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

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

64

65

locale.setlocale(locale.LC_ALL, "")

66

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

81

82

def main():

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

try:

97

bus = dbus.SystemBus()

98

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

99

dbus_busname, server_dbus_path)

100

mandos_dbus_objc = bus.get_object(dbus_busname,

101

server_dbus_path)

102

except dbus.exceptions.DBusException:

103

log.critical("Could not connect to Mandos server")

104

sys.exit(1)

105

106

mandos_serv = dbus.Interface(mandos_dbus_objc,

107

dbus_interface=server_dbus_interface)

108

mandos_serv_object_manager = dbus.Interface(

109

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

110

111

# Filter out log message from dbus module

112

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

113

class NullFilter(logging.Filter):

114

def filter(self, record):

115

return False

116

dbus_filter = NullFilter()

117

try:

118

dbus_logger.addFilter(dbus_filter)

119

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

120

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

121

all_clients = {path: ifs_and_props[client_dbus_interface]

122

for path, ifs_and_props in

123

mandos_serv_object_manager

124

.GetManagedObjects().items()

125

if client_dbus_interface in ifs_and_props}

126

except dbus.exceptions.DBusException as e:

127

log.critical("Failed to access Mandos server through D-Bus:"

128

"\n%s", e)

129

sys.exit(1)

130

finally:

131

# restore dbus logger

132

dbus_logger.removeFilter(dbus_filter)

133

134

# Compile dict of (clients: properties) to process

135

clients = {}

136

137

if not clientnames:

138

clients = all_clients

139

else:

140

for name in clientnames:

141

for objpath, properties in all_clients.items():

142

if properties["Name"] == name:

143

clients[objpath] = properties

144

break

145

else:

146

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

147

sys.exit(1)

148

149

# Run all commands on clients

150

commands = commands_from_options(options)

151

for command in commands:

152

command.run(clients, bus, mandos_serv)

153

154

155

def add_command_line_options(parser):

156

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

157

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

158

help="show version number and exit")

159

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

160

help="Select all clients")

161

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

162

help="Print all fields")

163

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

164

help="Dump client data in JSON format")

165

enable_disable = parser.add_mutually_exclusive_group()

166

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

167

help="Enable client")

168

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

169

action="store_true",

170

help="disable client")

171

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

172

help="Bump timeout for client")

173

start_stop_checker = parser.add_mutually_exclusive_group()

174

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

175

action="store_true",

176

help="Start checker for client")

177

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

178

action="store_true",

179

help="Stop checker for client")

180

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

181

help="Check if client is enabled")

182

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

183

help="Remove client")

184

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

185

help="Set checker command for client")

186

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

187

help="Set timeout for client")

188

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

189

help="Set extended timeout for client")

190

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

191

help="Set checker interval for client")

192

approve_deny_default = parser.add_mutually_exclusive_group()

193

approve_deny_default.add_argument(

194

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

195

default=None, dest="approved_by_default",

196

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

197

approve_deny_default.add_argument(

198

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

199

dest="approved_by_default",

200

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

201

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

202

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

203

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

204

help="Set duration of one client approval")

205

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

206

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

207

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

208

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

209

approve_deny = parser.add_mutually_exclusive_group()

210

approve_deny.add_argument(

211

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

212

help="Approve any current client request")

213

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

214

help="Deny any current client request")

215

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

216

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

217

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

218

help="Run self-test")

219

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

220

221

222

def string_to_delta(interval):

223

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

224

225

try:

226

return rfc3339_duration_to_delta(interval)

227

except ValueError as e:

228

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

229

' '.join(e.args))

230

return parse_pre_1_6_1_interval(interval)

231

232

233

def rfc3339_duration_to_delta(duration):

234

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

235

236

>>> rfc3339_duration_to_delta("P7D")

237

datetime.timedelta(7)

238

>>> rfc3339_duration_to_delta("PT60S")

239

datetime.timedelta(0, 60)

240

>>> rfc3339_duration_to_delta("PT60M")

241

datetime.timedelta(0, 3600)

242

>>> rfc3339_duration_to_delta("P60M")

243

datetime.timedelta(1680)

244

>>> rfc3339_duration_to_delta("PT24H")

245

datetime.timedelta(1)

246

>>> rfc3339_duration_to_delta("P1W")

247

datetime.timedelta(7)

248

>>> rfc3339_duration_to_delta("PT5M30S")

249

datetime.timedelta(0, 330)

250

>>> rfc3339_duration_to_delta("P1DT3M20S")

251

datetime.timedelta(1, 200)

252

>>> # Can not be empty:

253

>>> rfc3339_duration_to_delta("")

254

Traceback (most recent call last):

255

...

256

ValueError: Invalid RFC 3339 duration: u''

257

>>> # Must start with "P":

258

>>> rfc3339_duration_to_delta("1D")

259

Traceback (most recent call last):

260

...

261

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

262

>>> # Must use correct order

263

>>> rfc3339_duration_to_delta("PT1S2M")

264

Traceback (most recent call last):

265

...

266

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

267

>>> # Time needs time marker

268

>>> rfc3339_duration_to_delta("P1H2S")

269

Traceback (most recent call last):

270

...

271

ValueError: Invalid RFC 3339 duration: u'P1H2S'

272

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

273

>>> rfc3339_duration_to_delta("P1D2W")

274

Traceback (most recent call last):

275

...

276

ValueError: Invalid RFC 3339 duration: u'P1D2W'

277

>>> rfc3339_duration_to_delta("P2W2H")

278

Traceback (most recent call last):

279

...

280

ValueError: Invalid RFC 3339 duration: u'P2W2H'

281

"""

282

283

# Parsing an RFC 3339 duration with regular expressions is not

284

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

285

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

286

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

287

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

288

# avoid excessive use of external libraries.

289

290

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

291

Token = collections.namedtuple("Token", (

292

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

293

# a "group" containing digits

294

"value", # datetime.timedelta or None

295

"followers")) # Tokens valid after this token

296

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

297

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

298

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

299

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

300

datetime.timedelta(seconds=1),

301

frozenset((token_end, )))

302

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

303

datetime.timedelta(minutes=1),

304

frozenset((token_second, token_end)))

305

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

306

datetime.timedelta(hours=1),

307

frozenset((token_minute, token_end)))

308

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

309

None,

310

frozenset((token_hour, token_minute,

311

token_second)))

312

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

313

datetime.timedelta(days=1),

314

frozenset((token_time, token_end)))

315

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

316

datetime.timedelta(weeks=4),

317

frozenset((token_day, token_end)))

318

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

319

datetime.timedelta(weeks=52),

320

frozenset((token_month, token_end)))

321

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

322

datetime.timedelta(weeks=1),

323

frozenset((token_end, )))

324

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

325

frozenset((token_year, token_month,

326

token_day, token_time,

327

token_week)))

328

# Define starting values:

329

# Value so far

330

value = datetime.timedelta()

331

found_token = None

332

# Following valid tokens

333

followers = frozenset((token_duration, ))

334

# String left to parse

335

s = duration

336

# Loop until end token is found

337

while found_token is not token_end:

338

# Search for any currently valid tokens

339

for token in followers:

340

match = token.regexp.match(s)

341

if match is not None:

342

# Token found

343

if token.value is not None:

344

# Value found, parse digits

345

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

346

# Add to value so far

347

value += factor * token.value

348

# Strip token from string

349

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

350

# Go to found token

351

found_token = token

352

# Set valid next tokens

353

followers = found_token.followers

354

break

355

else:

356

# No currently valid tokens were found

357

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

358

.format(duration))

359

# End token found

360

return value

361

362

363

def parse_pre_1_6_1_interval(interval):

364

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

365

and return a datetime.timedelta

366

367

>>> parse_pre_1_6_1_interval('7d')

368

datetime.timedelta(7)

369

>>> parse_pre_1_6_1_interval('60s')

370

datetime.timedelta(0, 60)

371

>>> parse_pre_1_6_1_interval('60m')

372

datetime.timedelta(0, 3600)

373

>>> parse_pre_1_6_1_interval('24h')

374

datetime.timedelta(1)

375

>>> parse_pre_1_6_1_interval('1w')

376

datetime.timedelta(7)

377

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

378

datetime.timedelta(0, 330)

379

>>> parse_pre_1_6_1_interval('')

380

datetime.timedelta(0)

381

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

382

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

383

datetime.timedelta(2, 480, 18000)

384

385

"""

386

387

value = datetime.timedelta(0)

388

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

389

390

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

391

if suffix == "d":

392

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

393

elif suffix == "s":

394

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

395

elif suffix == "m":

396

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

397

elif suffix == "h":

398

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

399

elif suffix == "w":

400

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

401

elif suffix == "":

402

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

403

return value

404

405

406

def check_option_syntax(parser, options):

407

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

408

argparse"""

409

410

def has_actions(options):

411

return any((options.enable,

412

options.disable,

413

options.bump_timeout,

414

options.start_checker,

415

options.stop_checker,

416

options.is_enabled,

417

options.remove,

418

options.checker is not None,

419

options.timeout is not None,

420

options.extended_timeout is not None,

421

options.interval is not None,

422

options.approved_by_default is not None,

423

options.approval_delay is not None,

424

options.approval_duration is not None,

425

options.host is not None,

426

options.secret is not None,

427

options.approve,

428

options.deny))

429

430

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

431

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

432

if options.verbose and has_actions(options):

433

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

434

if options.dump_json and (options.verbose

435

or has_actions(options)):

436

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

437

if options.all and not has_actions(options):

438

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

439

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

440

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

441

if options.remove:

442

options.remove = False

443

if has_actions(options) and not options.deny:

444

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

445

options.remove = True

446

447

448

def commands_from_options(options):

449

450

commands = []

451

452

if options.is_enabled:

453

commands.append(IsEnabledCmd())

454

455

if options.approve:

456

commands.append(ApproveCmd())

457

458

if options.deny:

459

commands.append(DenyCmd())

460

461

if options.remove:

462

commands.append(RemoveCmd())

463

464

if options.dump_json:

465

commands.append(DumpJSONCmd())

466

467

if options.enable:

468

commands.append(EnableCmd())

469

470

if options.disable:

471

commands.append(DisableCmd())

472

473

if options.bump_timeout:

474

commands.append(BumpTimeoutCmd())

475

476

if options.start_checker:

477

commands.append(StartCheckerCmd())

478

479

if options.stop_checker:

480

commands.append(StopCheckerCmd())

481

482

if options.approved_by_default is not None:

483

if options.approved_by_default:

484

commands.append(ApproveByDefaultCmd())

485

else:

486

commands.append(DenyByDefaultCmd())

487

488

if options.checker is not None:

489

commands.append(SetCheckerCmd(options.checker))

490

491

if options.host is not None:

492

commands.append(SetHostCmd(options.host))

493

494

if options.secret is not None:

495

commands.append(SetSecretCmd(options.secret))

496

497

if options.timeout is not None:

498

commands.append(SetTimeoutCmd(options.timeout))

499

500

if options.extended_timeout:

501

commands.append(

502

SetExtendedTimeoutCmd(options.extended_timeout))

503

504

if options.interval is not None:

505

commands.append(SetIntervalCmd(options.interval))

506

507

if options.approval_delay is not None:

508

commands.append(SetApprovalDelayCmd(options.approval_delay))

509

510

if options.approval_duration is not None:

511

commands.append(

512

SetApprovalDurationCmd(options.approval_duration))

513

514

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

515

# optionally verbosely

516

if not commands:

517

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

518

519

return commands

520

521

522

class Command(object):

523

"""Abstract class for commands"""

524

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

525

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

526

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

527

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

528

self.mandos = mandos

529

for clientpath, properties in clients.items():

530

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

531

dbus_busname, str(clientpath))

532

client = bus.get_object(dbus_busname, clientpath)

533

self.run_on_one_client(client, properties)

534

535

536

class IsEnabledCmd(Command):

537

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

538

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

539

if self.is_enabled(client, properties):

540

sys.exit(0)

541

sys.exit(1)

542

def is_enabled(self, client, properties):

543

return properties["Enabled"]

544

545

546

class ApproveCmd(Command):

547

def run_on_one_client(self, client, properties):

548

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

549

client.__dbus_object_path__, client_dbus_interface)

550

client.Approve(dbus.Boolean(True),

551

dbus_interface=client_dbus_interface)

552

553

554

class DenyCmd(Command):

555

def run_on_one_client(self, client, properties):

556

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

557

client.__dbus_object_path__, client_dbus_interface)

558

client.Approve(dbus.Boolean(False),

559

dbus_interface=client_dbus_interface)

560

561

562

class RemoveCmd(Command):

563

def run_on_one_client(self, client, properties):

564

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

565

server_dbus_path, server_dbus_interface,

566

str(client.__dbus_object_path__))

567

self.mandos.RemoveClient(client.__dbus_object_path__)

568

569

570

class OutputCmd(Command):

571

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

572

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

573

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

574

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

575

"ApprovalPending", "ApprovedByDefault",

576

"LastApprovalRequest", "ApprovalDelay",

577

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

578

"Expires", "LastCheckerStatus")

579

580

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

581

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

582

583

def output(self, clients):

584

raise NotImplementedError()

585

586

587

class DumpJSONCmd(OutputCmd):

588

def output(self, clients):

589

data = {client["Name"]:

590

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

591

for key in self.all_keywords}

592

for client in clients}

593

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

594

595

@staticmethod

596

def dbus_boolean_to_bool(value):

597

if isinstance(value, dbus.Boolean):

598

value = bool(value)

599

return value

600

601

602

class PrintTableCmd(OutputCmd):

603

def __init__(self, verbose=False):

604

self.verbose = verbose

605

606

def output(self, clients):

607

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

608

"LastCheckedOK")

609

keywords = default_keywords

610

if self.verbose:

611

keywords = self.all_keywords

612

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

613

614

class TableOfClients(object):

615

tableheaders = {

616

"Name": "Name",

617

"Enabled": "Enabled",

618

"Timeout": "Timeout",

619

"LastCheckedOK": "Last Successful Check",

620

"LastApprovalRequest": "Last Approval Request",

621

"Created": "Created",

622

"Interval": "Interval",

623

"Host": "Host",

624

"Fingerprint": "Fingerprint",

625

"KeyID": "Key ID",

626

"CheckerRunning": "Check Is Running",

627

"LastEnabled": "Last Enabled",

628

"ApprovalPending": "Approval Is Pending",

629

"ApprovedByDefault": "Approved By Default",

630

"ApprovalDelay": "Approval Delay",

631

"ApprovalDuration": "Approval Duration",

632

"Checker": "Checker",

633

"ExtendedTimeout": "Extended Timeout",

634

"Expires": "Expires",

635

"LastCheckerStatus": "Last Checker Status",

636

}

637

638

def __init__(self, clients, keywords):

639

self.clients = clients

640

self.keywords = keywords

641

642

def __str__(self):

643

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

644

645

if sys.version_info.major == 2:

646

__unicode__ = __str__

647

def __str__(self):

648

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

649

650

def rows(self):

651

format_string = self.row_formatting_string()

652

rows = [self.header_line(format_string)]

653

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

654

for client in self.clients)

655

return rows

656

657

def row_formatting_string(self):

658

"Format string used to format table rows"

659

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

660

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

661

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

662

for client in self.clients)),

663

key=key)

664

for key in self.keywords)

665

666

def string_from_client(self, client, key):

667

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

668

669

@classmethod

670

def valuetostring(cls, value, keyword):

671

if isinstance(value, dbus.Boolean):

672

return "Yes" if value else "No"

673

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

674

"ApprovalDuration", "ExtendedTimeout"):

675

return cls.milliseconds_to_string(value)

676

return str(value)

677

678

def header_line(self, format_string):

679

return format_string.format(**self.tableheaders)

680

681

def client_line(self, client, format_string):

682

return format_string.format(

683

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

684

for key in self.keywords})

685

686

@staticmethod

687

def milliseconds_to_string(ms):

688

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

689

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

690

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

691

if td.days else "",

692

hours=td.seconds // 3600,

693

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

694

seconds=td.seconds % 60))

695

696

697

class PropertyCmd(Command):

698

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

699

700

def run_on_one_client(self, client, properties):

701

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

702

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

703

client.__dbus_object_path__,

704

dbus.PROPERTIES_IFACE, client_dbus_interface,

705

self.propname, self.value_to_set

706

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

707

else bool(self.value_to_set))

708

client.Set(client_dbus_interface, self.propname,

709

self.value_to_set,

710

dbus_interface=dbus.PROPERTIES_IFACE)

711

712

@property

713

def propname(self):

714

raise NotImplementedError()

715

716

717

class EnableCmd(PropertyCmd):

718

propname = "Enabled"

719

value_to_set = dbus.Boolean(True)

720

721

722

class DisableCmd(PropertyCmd):

723

propname = "Enabled"

724

value_to_set = dbus.Boolean(False)

725

726

727

class BumpTimeoutCmd(PropertyCmd):

728

propname = "LastCheckedOK"

729

value_to_set = ""

730

731

732

class StartCheckerCmd(PropertyCmd):

733

propname = "CheckerRunning"

734

value_to_set = dbus.Boolean(True)

735

736

737

class StopCheckerCmd(PropertyCmd):

738

propname = "CheckerRunning"

739

value_to_set = dbus.Boolean(False)

740

741

742

class ApproveByDefaultCmd(PropertyCmd):

743

propname = "ApprovedByDefault"

744

value_to_set = dbus.Boolean(True)

745

746

747

class DenyByDefaultCmd(PropertyCmd):

748

propname = "ApprovedByDefault"

749

value_to_set = dbus.Boolean(False)

750

751

752

class PropertyValueCmd(PropertyCmd):

753

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

754

def __init__(self, value):

755

self.value_to_set = value

756

757

758

class SetCheckerCmd(PropertyValueCmd):

759

propname = "Checker"

760

761

762

class SetHostCmd(PropertyValueCmd):

763

propname = "Host"

764

765

766

class SetSecretCmd(PropertyValueCmd):

767

propname = "Secret"

768

769

@property

770

def value_to_set(self):

771

return self._vts

772

773

@value_to_set.setter

774

def value_to_set(self, value):

775

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

776

self._vts = value.read()

777

value.close()

778

779

780

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

781

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

782

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

783

784

@property

785

def value_to_set(self):

786

return self._vts

787

788

@value_to_set.setter

789

def value_to_set(self, value):

790

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

791

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

792

793

794

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

795

propname = "Timeout"

796

797

798

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

799

propname = "ExtendedTimeout"

800

801

802

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

803

propname = "Interval"

804

805

806

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

807

propname = "ApprovalDelay"

808

809

810

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

811

propname = "ApprovalDuration"

812

813

814

815

class Test_string_to_delta(unittest.TestCase):

816

def test_handles_basic_rfc3339(self):

817

self.assertEqual(string_to_delta("PT0S"),

818

datetime.timedelta())

819

self.assertEqual(string_to_delta("P0D"),

820

datetime.timedelta())

821

self.assertEqual(string_to_delta("PT1S"),

822

datetime.timedelta(0, 1))

823

self.assertEqual(string_to_delta("PT2H"),

824

datetime.timedelta(0, 7200))

825

826

def test_falls_back_to_pre_1_6_1_with_warning(self):

827

# assertLogs only exists in Python 3.4

828

if hasattr(self, "assertLogs"):

829

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

830

value = string_to_delta("2h")

831

else:

832

class WarningFilter(logging.Filter):

833

"""Don't show, but record the presence of, warnings"""

834

def filter(self, record):

835

is_warning = record.levelno >= logging.WARNING

836

self.found = is_warning or getattr(self, "found",

837

False)

838

return not is_warning

839

warning_filter = WarningFilter()

840

log.addFilter(warning_filter)

841

try:

842

value = string_to_delta("2h")

843

finally:

844

log.removeFilter(warning_filter)

845

self.assertTrue(getattr(warning_filter, "found", False))

846

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

847

848

849

class Test_check_option_syntax(unittest.TestCase):

850

def setUp(self):

851

self.parser = argparse.ArgumentParser()

852

add_command_line_options(self.parser)

853

854

def test_actions_requires_client_or_all(self):

855

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

856

options = self.parser.parse_args()

857

setattr(options, action, value)

858

with self.assertParseError():

859

self.check_option_syntax(options)

860

861

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

862

# check_option_syntax()

863

actions = {

864

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

865

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

866

# never used

867

"enable": True,

868

"disable": True,

869

"bump_timeout": True,

870

"start_checker": True,

871

"stop_checker": True,

872

"is_enabled": True,

873

"remove": True,

874

"checker": "x",

875

"timeout": datetime.timedelta(),

876

"extended_timeout": datetime.timedelta(),

877

"interval": datetime.timedelta(),

878

"approved_by_default": True,

879

"approval_delay": datetime.timedelta(),

880

"approval_duration": datetime.timedelta(),

881

"host": "x",

882

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

883

"approve": True,

884

"deny": True,

885

}

886

887

@contextlib.contextmanager

888

def assertParseError(self):

889

with self.assertRaises(SystemExit) as e:

890

with self.temporarily_suppress_stderr():

891

yield

892

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

893

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

894

# /argparse.html#exiting-methods

895

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

896

897

@staticmethod

898

@contextlib.contextmanager

899

def temporarily_suppress_stderr():

900

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

901

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

902

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

903

os.close(null)

904

try:

905

yield

906

finally:

907

# restore stderr

908

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

909

os.close(stderrcopy)

910

911

def check_option_syntax(self, options):

912

check_option_syntax(self.parser, options)

913

914

def test_actions_conflicts_with_verbose(self):

915

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

916

options = self.parser.parse_args()

917

setattr(options, action, value)

918

options.verbose = True

919

with self.assertParseError():

920

self.check_option_syntax(options)

921

922

def test_dump_json_conflicts_with_verbose(self):

923

options = self.parser.parse_args()

924

options.dump_json = True

925

options.verbose = True

926

with self.assertParseError():

927

self.check_option_syntax(options)

928

929

def test_dump_json_conflicts_with_action(self):

930

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

931

options = self.parser.parse_args()

932

setattr(options, action, value)

933

options.dump_json = True

934

with self.assertParseError():

935

self.check_option_syntax(options)

936

937

def test_all_can_not_be_alone(self):

938

options = self.parser.parse_args()

939

options.all = True

940

with self.assertParseError():

941

self.check_option_syntax(options)

942

943

def test_all_is_ok_with_any_action(self):

944

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

945

options = self.parser.parse_args()

946

setattr(options, action, value)

947

options.all = True

948

self.check_option_syntax(options)

949

950

def test_is_enabled_fails_without_client(self):

951

options = self.parser.parse_args()

952

options.is_enabled = True

953

with self.assertParseError():

954

self.check_option_syntax(options)

955

956

def test_is_enabled_works_with_one_client(self):

957

options = self.parser.parse_args()

958

options.is_enabled = True

959

options.client = ["foo"]

960

self.check_option_syntax(options)

961

962

def test_is_enabled_fails_with_two_clients(self):

963

options = self.parser.parse_args()

964

options.is_enabled = True

965

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

966

with self.assertParseError():

967

self.check_option_syntax(options)

968

969

def test_remove_can_only_be_combined_with_action_deny(self):

970

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

971

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

972

continue

973

options = self.parser.parse_args()

974

setattr(options, action, value)

975

options.all = True

976

options.remove = True

977

with self.assertParseError():

978

self.check_option_syntax(options)

979

980

981

class Test_commands_from_options(unittest.TestCase):

982

def setUp(self):

983

self.parser = argparse.ArgumentParser()

984

add_command_line_options(self.parser)

985

986

def test_is_enabled(self):

987

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

988

IsEnabledCmd)

989

990

def assert_command_from_args(self, args, command_cls,

991

**cmd_attrs):

992

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

993

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

994

options = self.parser.parse_args(args)

995

check_option_syntax(self.parser, options)

996

commands = commands_from_options(options)

997

self.assertEqual(len(commands), 1)

998

command = commands[0]

999

self.assertIsInstance(command, command_cls)

1000

for key, value in cmd_attrs.items():

1001

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

1002

1003

def test_is_enabled_short(self):

1004

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

1005

1006

def test_approve(self):

1007

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

1008

ApproveCmd)

1009

1010

def test_approve_short(self):

1011

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

1012

1013

def test_deny(self):

1014

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

1015

1016

def test_deny_short(self):

1017

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

1018

1019

def test_remove(self):

1020

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

1021

RemoveCmd)

1022

1023

def test_deny_before_remove(self):

1024

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

1025

"foo"])

1026

check_option_syntax(self.parser, options)

1027

commands = commands_from_options(options)

1028

self.assertEqual(len(commands), 2)

1029

self.assertIsInstance(commands[0], DenyCmd)

1030

self.assertIsInstance(commands[1], RemoveCmd)

1031

1032

def test_deny_before_remove_reversed(self):

1033

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

1034

"--all"])

1035

check_option_syntax(self.parser, options)

1036

commands = commands_from_options(options)

1037

self.assertEqual(len(commands), 2)

1038

self.assertIsInstance(commands[0], DenyCmd)

1039

self.assertIsInstance(commands[1], RemoveCmd)

1040

1041

def test_remove_short(self):

1042

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

1043

1044

def test_dump_json(self):

1045

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

1046

1047

def test_enable(self):

1048

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

1049

1050

def test_enable_short(self):

1051

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

1052

1053

def test_disable(self):

1054

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

1055

DisableCmd)

1056

1057

def test_disable_short(self):

1058

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

1059

1060

def test_bump_timeout(self):

1061

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

1062

BumpTimeoutCmd)

1063

1064

def test_bump_timeout_short(self):

1065

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

1066

1067

def test_start_checker(self):

1068

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

1069

StartCheckerCmd)

1070

1071

def test_stop_checker(self):

1072

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

1073

StopCheckerCmd)

1074

1075

def test_approve_by_default(self):

1076

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

1077

ApproveByDefaultCmd)

1078

1079

def test_deny_by_default(self):

1080

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

1081

DenyByDefaultCmd)

1082

1083

def test_checker(self):

1084

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

1085

SetCheckerCmd, value_to_set=":")

1086

1087

def test_checker_empty(self):

1088

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

1089

SetCheckerCmd, value_to_set="")

1090

1091

def test_checker_short(self):

1092

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

1093

SetCheckerCmd, value_to_set=":")

1094

1095

def test_host(self):

1096

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

1097

"foo"], SetHostCmd,

1098

value_to_set="foo.example.org")

1099

1100

def test_host_short(self):

1101

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

1102

"foo"], SetHostCmd,

1103

value_to_set="foo.example.org")

1104

1105

def test_secret_devnull(self):

1106

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

1107

"foo"], SetSecretCmd,

1108

value_to_set=b"")

1109

1110

def test_secret_tempfile(self):

1111

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

1112

value = b"secret\0xyzzy\nbar"

1113

f.write(value)

1114

f.seek(0)

1115

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

1116

"foo"], SetSecretCmd,

1117

value_to_set=value)

1118

1119

def test_secret_devnull_short(self):

1120

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

1121

SetSecretCmd, value_to_set=b"")

1122

1123

def test_secret_tempfile_short(self):

1124

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

1125

value = b"secret\0xyzzy\nbar"

1126

f.write(value)

1127

f.seek(0)

1128

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

1129

SetSecretCmd,

1130

value_to_set=value)

1131

1132

def test_timeout(self):

1133

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

1134

SetTimeoutCmd,

1135

value_to_set=300000)

1136

1137

def test_timeout_short(self):

1138

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

1139

SetTimeoutCmd,

1140

value_to_set=300000)

1141

1142

def test_extended_timeout(self):

1143

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

1144

"foo"],

1145

SetExtendedTimeoutCmd,

1146

value_to_set=900000)

1147

1148

def test_interval(self):

1149

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

1150

SetIntervalCmd,

1151

value_to_set=120000)

1152

1153

def test_interval_short(self):

1154

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

1155

SetIntervalCmd,

1156

value_to_set=120000)

1157

1158

def test_approval_delay(self):

1159

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

1160

"foo"], SetApprovalDelayCmd,

1161

value_to_set=30000)

1162

1163

def test_approval_duration(self):

1164

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

1165

"foo"], SetApprovalDurationCmd,

1166

value_to_set=1000)

1167

1168

def test_print_table(self):

1169

self.assert_command_from_args([], PrintTableCmd,

1170

verbose=False)

1171

1172

def test_print_table_verbose(self):

1173

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

1174

verbose=True)

1175

1176

def test_print_table_verbose_short(self):

1177

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

1178

verbose=True)

1179

1180

1181

class TestCmd(unittest.TestCase):

1182

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

1183

1184

def setUp(self):

1185

testcase = self

1186

class MockClient(object):

1187

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

1188

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

1189

self.attributes = attributes

1190

self.attributes["Name"] = name

1191

self.calls = []

1192

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

1193

testcase.assertEqual(interface, client_dbus_interface)

1194

testcase.assertEqual(dbus_interface,

1195

dbus.PROPERTIES_IFACE)

1196

self.attributes[propname] = value

1197

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

1198

testcase.assertEqual(interface, client_dbus_interface)

1199

testcase.assertEqual(dbus_interface,

1200

dbus.PROPERTIES_IFACE)

1201

return self.attributes[propname]

1202

def Approve(self, approve, dbus_interface):

1203

testcase.assertEqual(dbus_interface,

1204

client_dbus_interface)

1205

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

1206

dbus_interface)))

1207

self.client = MockClient(

1208

"foo",

1209

KeyID=("92ed150794387c03ce684574b1139a65"

1210

"94a34f895daaaf09fd8ea90a27cddb12"),

1211

Secret=b"secret",

1212

Host="foo.example.org",

1213

Enabled=dbus.Boolean(True),

1214

Timeout=300000,

1215

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

1216

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

1217

Interval=120000,

1218

Fingerprint=("778827225BA7DE539C5A"

1219

"7CFA59CFF7CDBD9A5920"),

1220

CheckerRunning=dbus.Boolean(False),

1221

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

1222

ApprovalPending=dbus.Boolean(False),

1223

ApprovedByDefault=dbus.Boolean(True),

1224

LastApprovalRequest="",

1225

ApprovalDelay=0,

1226

ApprovalDuration=1000,

1227

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

1228

ExtendedTimeout=900000,

1229

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

1230

LastCheckerStatus=0)

1231

self.other_client = MockClient(

1232

"barbar",

1233

KeyID=("0558568eedd67d622f5c83b35a115f79"

1234

"6ab612cff5ad227247e46c2b020f441c"),

1235

Secret=b"secretbar",

1236

Host="192.0.2.3",

1237

Enabled=dbus.Boolean(True),

1238

Timeout=300000,

1239

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

1240

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

1241

Interval=120000,

1242

Fingerprint=("3E393AEAEFB84C7E89E2"

1243

"F547B3A107558FCA3A27"),

1244

CheckerRunning=dbus.Boolean(True),

1245

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

1246

ApprovalPending=dbus.Boolean(False),

1247

ApprovedByDefault=dbus.Boolean(False),

1248

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

1249

ApprovalDelay=30000,

1250

ApprovalDuration=93785000,

1251

Checker=":",

1252

ExtendedTimeout=900000,

1253

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

1254

LastCheckerStatus=-2)

1255

self.clients = collections.OrderedDict(

1256

[

1257

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

1258

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

1259

])

1260

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

1261

1262

@property

1263

def bus(self):

1264

class Bus(object):

1265

@staticmethod

1266

def get_object(client_bus_name, path):

1267

self.assertEqual(client_bus_name, dbus_busname)

1268

return {

1269

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

1270

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

1271

"/clients/foo": self.client,

1272

"/clients/barbar": self.other_client,

1273

}[path]

1274

return Bus()

1275

1276

1277

class TestIsEnabledCmd(TestCmd):

1278

def test_is_enabled(self):

1279

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

1280

properties)

1281

for client, properties

1282

in self.clients.items()))

1283

1284

def test_is_enabled_run_exits_successfully(self):

1285

with self.assertRaises(SystemExit) as e:

1286

IsEnabledCmd().run(self.one_client)

1287

if e.exception.code is not None:

1288

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

1289

else:

1290

self.assertIsNone(e.exception.code)

1291

1292

def test_is_enabled_run_exits_with_failure(self):

1293

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

1294

with self.assertRaises(SystemExit) as e:

1295

IsEnabledCmd().run(self.one_client)

1296

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

1297

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

1298

else:

1299

self.assertIsNotNone(e.exception.code)

1300

1301

1302

class TestApproveCmd(TestCmd):

1303

def test_approve(self):

1304

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

1305

for clientpath in self.clients:

1306

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

1307

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

1308

client.calls)

1309

1310

1311

class TestDenyCmd(TestCmd):

1312

def test_deny(self):

1313

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

1314

for clientpath in self.clients:

1315

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

1316

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

1317

client.calls)

1318

1319

1320

class TestRemoveCmd(TestCmd):

1321

def test_remove(self):

1322

class MockMandos(object):

1323

def __init__(self):

1324

self.calls = []

1325

def RemoveClient(self, dbus_path):

1326

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

1327

mandos = MockMandos()

1328

super(TestRemoveCmd, self).setUp()

1329

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

1330

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

1331

for clientpath in self.clients:

1332

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

1333

mandos.calls)

1334

1335

1336

class TestDumpJSONCmd(TestCmd):

1337

def setUp(self):

1338

self.expected_json = {

1339

"foo": {

1340

"Name": "foo",

1341

"KeyID": ("92ed150794387c03ce684574b1139a65"

1342

"94a34f895daaaf09fd8ea90a27cddb12"),

1343

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

1344

"Enabled": True,

1345

"Timeout": 300000,

1346

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

1347

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

1348

"Interval": 120000,

1349

"Fingerprint": ("778827225BA7DE539C5A"

1350

"7CFA59CFF7CDBD9A5920"),

1351

"CheckerRunning": False,

1352

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

1353

"ApprovalPending": False,

1354

"ApprovedByDefault": True,

1355

"LastApprovalRequest": "",

1356

"ApprovalDelay": 0,

1357

"ApprovalDuration": 1000,

1358

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

1359

"ExtendedTimeout": 900000,

1360

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

1361

"LastCheckerStatus": 0,

1362

},

1363

"barbar": {

1364

"Name": "barbar",

1365

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1366

"6ab612cff5ad227247e46c2b020f441c"),

1367

"Host": "192.0.2.3",

1368

"Enabled": True,

1369

"Timeout": 300000,

1370

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

1371

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

1372

"Interval": 120000,

1373

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1374

"F547B3A107558FCA3A27"),

1375

"CheckerRunning": True,

1376

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

1377

"ApprovalPending": False,

1378

"ApprovedByDefault": False,

1379

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

1380

"ApprovalDelay": 30000,

1381

"ApprovalDuration": 93785000,

1382

"Checker": ":",

1383

"ExtendedTimeout": 900000,

1384

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

1385

"LastCheckerStatus": -2,

1386

},

1387

}

1388

return super(TestDumpJSONCmd, self).setUp()

1389

1390

def test_normal(self):

1391

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

1392

json_data = json.loads(output)

1393

self.assertDictEqual(json_data, self.expected_json)

1394

1395

def test_one_client(self):

1396

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

1397

json_data = json.loads(output)

1398

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

1399

self.assertDictEqual(json_data, expected_json)

1400

1401

1402

class TestPrintTableCmd(TestCmd):

1403

def test_normal(self):

1404

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

1405

expected_output = "\n".join((

1406

"Name Enabled Timeout Last Successful Check",

1407

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

1408

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

1409

))

1410

self.assertEqual(output, expected_output)

1411

1412

def test_verbose(self):

1413

output = PrintTableCmd(verbose=True).output(

1414

self.clients.values())

1415

columns = (

1416

(

1417

"Name ",

1418

"foo ",

1419

"barbar ",

1420

),(

1421

"Enabled ",

1422

"Yes ",

1423

"Yes ",

1424

),(

1425

"Timeout ",

1426

"00:05:00 ",

1427

"00:05:00 ",

1428

),(

1429

"Last Successful Check ",

1430

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

1431

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

1432

),(

1433

"Created ",

1434

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

1435

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

1436

),(

1437

"Interval ",

1438

"00:02:00 ",

1439

"00:02:00 ",

1440

),(

1441

"Host ",

1442

"foo.example.org ",

1443

"192.0.2.3 ",

1444

),(

1445

("Key ID "

1446

" "),

1447

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1448

"ea90a27cddb12 "),

1449

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1450

"46c2b020f441c "),

1451

),(

1452

"Fingerprint ",

1453

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1454

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1455

),(

1456

"Check Is Running ",

1457

"No ",

1458

"Yes ",

1459

),(

1460

"Last Enabled ",

1461

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

1462

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

1463

),(

1464

"Approval Is Pending ",

1465

"No ",

1466

"No ",

1467

),(

1468

"Approved By Default ",

1469

"Yes ",

1470

"No ",

1471

),(

1472

"Last Approval Request ",

1473

" ",

1474

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

1475

),(

1476

"Approval Delay ",

1477

"00:00:00 ",

1478

"00:00:30 ",

1479

),(

1480

"Approval Duration ",

1481

"00:00:01 ",

1482

"1T02:03:05 ",

1483

),(

1484

"Checker ",

1485

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

1486

": ",

1487

),(

1488

"Extended Timeout ",

1489

"00:15:00 ",

1490

"00:15:00 ",

1491

),(

1492

"Expires ",

1493

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

1494

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

1495

),(

1496

"Last Checker Status",

1497

"0 ",

1498

"-2 ",

1499

)

1500

)

1501

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

1502

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

1503

for rows in columns)

1504

for line in range(num_lines))

1505

self.assertEqual(output, expected_output)

1506

1507

def test_one_client(self):

1508

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

1509

expected_output = "\n".join((

1510

"Name Enabled Timeout Last Successful Check",

1511

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

1512

))

1513

self.assertEqual(output, expected_output)

1514

1515

1516

class TestPropertyCmd(TestCmd):

1517

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

1518

def runTest(self):

1519

if not hasattr(self, "command"):

1520

return

1521

values_to_get = getattr(self, "values_to_get",

1522

self.values_to_set)

1523

for value_to_set, value_to_get in zip(self.values_to_set,

1524

values_to_get):

1525

for clientpath in self.clients:

1526

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

1527

old_value = client.attributes[self.propname]

1528

self.assertNotIsInstance(old_value, self.Unique)

1529

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

1530

self.run_command(value_to_set, self.clients)

1531

for clientpath in self.clients:

1532

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

1533

value = client.attributes[self.propname]

1534

self.assertNotIsInstance(value, self.Unique)

1535

self.assertEqual(value, value_to_get)

1536

1537

class Unique(object):

1538

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

1539

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

1540

1541

def run_command(self, value, clients):

1542

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

1543

1544

1545

class TestEnableCmd(TestPropertyCmd):

1546

command = EnableCmd

1547

propname = "Enabled"

1548

values_to_set = [dbus.Boolean(True)]

1549

1550

1551

class TestDisableCmd(TestPropertyCmd):

1552

command = DisableCmd

1553

propname = "Enabled"

1554

values_to_set = [dbus.Boolean(False)]

1555

1556

1557

class TestBumpTimeoutCmd(TestPropertyCmd):

1558

command = BumpTimeoutCmd

1559

propname = "LastCheckedOK"

1560

values_to_set = [""]

1561

1562

1563

class TestStartCheckerCmd(TestPropertyCmd):

1564

command = StartCheckerCmd

1565

propname = "CheckerRunning"

1566

values_to_set = [dbus.Boolean(True)]

1567

1568

1569

class TestStopCheckerCmd(TestPropertyCmd):

1570

command = StopCheckerCmd

1571

propname = "CheckerRunning"

1572

values_to_set = [dbus.Boolean(False)]

1573

1574

1575

class TestApproveByDefaultCmd(TestPropertyCmd):

1576

command = ApproveByDefaultCmd

1577

propname = "ApprovedByDefault"

1578

values_to_set = [dbus.Boolean(True)]

1579

1580

1581

class TestDenyByDefaultCmd(TestPropertyCmd):

1582

command = DenyByDefaultCmd

1583

propname = "ApprovedByDefault"

1584

values_to_set = [dbus.Boolean(False)]

1585

1586

1587

class TestPropertyValueCmd(TestPropertyCmd):

1588

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

1589

1590

def runTest(self):

1591

if type(self) is TestPropertyValueCmd:

1592

return

1593

return super(TestPropertyValueCmd, self).runTest()

1594

1595

def run_command(self, value, clients):

1596

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

1597

1598

1599

class TestSetCheckerCmd(TestPropertyValueCmd):

1600

command = SetCheckerCmd

1601

propname = "Checker"

1602

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

1603

1604

1605

class TestSetHostCmd(TestPropertyValueCmd):

1606

command = SetHostCmd

1607

propname = "Host"

1608

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

1609

1610

1611

class TestSetSecretCmd(TestPropertyValueCmd):

1612

command = SetSecretCmd

1613

propname = "Secret"

1614

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

1615

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

1616

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

1617

1618

1619

class TestSetTimeoutCmd(TestPropertyValueCmd):

1620

command = SetTimeoutCmd

1621

propname = "Timeout"

1622

values_to_set = [datetime.timedelta(),

1623

datetime.timedelta(minutes=5),

1624

datetime.timedelta(seconds=1),

1625

datetime.timedelta(weeks=1),

1626

datetime.timedelta(weeks=52)]

1627

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

1628

1629

1630

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1631

command = SetExtendedTimeoutCmd

1632

propname = "ExtendedTimeout"

1633

values_to_set = [datetime.timedelta(),

1634

datetime.timedelta(minutes=5),

1635

datetime.timedelta(seconds=1),

1636

datetime.timedelta(weeks=1),

1637

datetime.timedelta(weeks=52)]

1638

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

1639

1640

1641

class TestSetIntervalCmd(TestPropertyValueCmd):

1642

command = SetIntervalCmd

1643

propname = "Interval"

1644

values_to_set = [datetime.timedelta(),

1645

datetime.timedelta(minutes=5),

1646

datetime.timedelta(seconds=1),

1647

datetime.timedelta(weeks=1),

1648

datetime.timedelta(weeks=52)]

1649

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

1650

1651

1652

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1653

command = SetApprovalDelayCmd

1654

propname = "ApprovalDelay"

1655

values_to_set = [datetime.timedelta(),

1656

datetime.timedelta(minutes=5),

1657

datetime.timedelta(seconds=1),

1658

datetime.timedelta(weeks=1),

1659

datetime.timedelta(weeks=52)]

1660

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

1661

1662

1663

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1664

command = SetApprovalDurationCmd

1665

propname = "ApprovalDuration"

1666

values_to_set = [datetime.timedelta(),

1667

datetime.timedelta(minutes=5),

1668

datetime.timedelta(seconds=1),

1669

datetime.timedelta(weeks=1),

1670

datetime.timedelta(weeks=52)]

1671

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

1672

1673

1674

1675

def should_only_run_tests():

1676

parser = argparse.ArgumentParser(add_help=False)

1677

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

1678

args, unknown_args = parser.parse_known_args()

1679

run_tests = args.check

1680

if run_tests:

1681

# Remove --check argument from sys.argv

1682

sys.argv[1:] = unknown_args

1683

return run_tests

1684

1685

# Add all tests from doctest strings

1686

def load_tests(loader, tests, none):

1687

import doctest

1688

tests.addTests(doctest.DocTestSuite())

1689

return tests

1690

1691

if __name__ == "__main__":

1692

if should_only_run_tests():

1693

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

1694

unittest.main()

1695

else:

1696

main()