/mandos/release : revision 237.7.612

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

mandos_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 = {objpath: properties

139

for objpath, properties in mandos_clients.items()}

140

else:

141

for name in clientnames:

142

for objpath, properties in mandos_clients.items():

143

if properties["Name"] == name:

144

clients[objpath] = properties

145

break

146

else:

147

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

148

sys.exit(1)

149

150

# Run all commands on clients

151

commands = commands_from_options(options)

152

for command in commands:

153

command.run(clients, bus, mandos_serv)

154

155

156

def add_command_line_options(parser):

157

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

158

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

159

help="show version number and exit")

160

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

161

help="Select all clients")

162

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

163

help="Print all fields")

164

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

165

help="Dump client data in JSON format")

166

enable_disable = parser.add_mutually_exclusive_group()

167

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

168

help="Enable client")

169

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

170

action="store_true",

171

help="disable client")

172

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

173

help="Bump timeout for client")

174

start_stop_checker = parser.add_mutually_exclusive_group()

175

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

176

action="store_true",

177

help="Start checker for client")

178

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

179

action="store_true",

180

help="Stop checker for client")

181

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

182

help="Check if client is enabled")

183

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

184

help="Remove client")

185

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

186

help="Set checker command for client")

187

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

188

help="Set timeout for client")

189

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

190

help="Set extended timeout for client")

191

parser.add_argument("-i", "--interval", type=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", action="store_true",

196

default=None, dest="approved_by_default",

197

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

198

approve_deny_default.add_argument(

199

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

200

dest="approved_by_default",

201

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

202

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

203

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

204

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

205

help="Set duration of one client approval")

206

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

207

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

208

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

209

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

210

approve_deny = parser.add_mutually_exclusive_group()

211

approve_deny.add_argument(

212

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

213

help="Approve any current client request")

214

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

215

help="Deny any current client request")

216

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

217

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

218

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

219

help="Run self-test")

220

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

221

222

223

def string_to_delta(interval):

224

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

225

226

try:

227

return rfc3339_duration_to_delta(interval)

228

except ValueError as e:

229

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

230

' '.join(e.args))

231

return parse_pre_1_6_1_interval(interval)

232

233

234

def rfc3339_duration_to_delta(duration):

235

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

236

237

>>> rfc3339_duration_to_delta("P7D")

238

datetime.timedelta(7)

239

>>> rfc3339_duration_to_delta("PT60S")

240

datetime.timedelta(0, 60)

241

>>> rfc3339_duration_to_delta("PT60M")

242

datetime.timedelta(0, 3600)

243

>>> rfc3339_duration_to_delta("P60M")

244

datetime.timedelta(1680)

245

>>> rfc3339_duration_to_delta("PT24H")

246

datetime.timedelta(1)

247

>>> rfc3339_duration_to_delta("P1W")

248

datetime.timedelta(7)

249

>>> rfc3339_duration_to_delta("PT5M30S")

250

datetime.timedelta(0, 330)

251

>>> rfc3339_duration_to_delta("P1DT3M20S")

252

datetime.timedelta(1, 200)

253

>>> # Can not be empty:

254

>>> rfc3339_duration_to_delta("")

255

Traceback (most recent call last):

256

...

257

ValueError: Invalid RFC 3339 duration: u''

258

>>> # Must start with "P":

259

>>> rfc3339_duration_to_delta("1D")

260

Traceback (most recent call last):

261

...

262

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

263

>>> # Must use correct order

264

>>> rfc3339_duration_to_delta("PT1S2M")

265

Traceback (most recent call last):

266

...

267

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

268

>>> # Time needs time marker

269

>>> rfc3339_duration_to_delta("P1H2S")

270

Traceback (most recent call last):

271

...

272

ValueError: Invalid RFC 3339 duration: u'P1H2S'

273

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

274

>>> rfc3339_duration_to_delta("P1D2W")

275

Traceback (most recent call last):

276

...

277

ValueError: Invalid RFC 3339 duration: u'P1D2W'

278

>>> rfc3339_duration_to_delta("P2W2H")

279

Traceback (most recent call last):

280

...

281

ValueError: Invalid RFC 3339 duration: u'P2W2H'

282

"""

283

284

# Parsing an RFC 3339 duration with regular expressions is not

285

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

286

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

287

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

288

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

289

# avoid excessive use of external libraries.

290

291

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

292

Token = collections.namedtuple("Token", (

293

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

294

# a "group" containing digits

295

"value", # datetime.timedelta or None

296

"followers")) # Tokens valid after this token

297

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

298

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

299

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

300

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

301

datetime.timedelta(seconds=1),

302

frozenset((token_end, )))

303

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

304

datetime.timedelta(minutes=1),

305

frozenset((token_second, token_end)))

306

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

307

datetime.timedelta(hours=1),

308

frozenset((token_minute, token_end)))

309

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

310

None,

311

frozenset((token_hour, token_minute,

312

token_second)))

313

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

314

datetime.timedelta(days=1),

315

frozenset((token_time, token_end)))

316

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

317

datetime.timedelta(weeks=4),

318

frozenset((token_day, token_end)))

319

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

320

datetime.timedelta(weeks=52),

321

frozenset((token_month, token_end)))

322

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

323

datetime.timedelta(weeks=1),

324

frozenset((token_end, )))

325

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

326

frozenset((token_year, token_month,

327

token_day, token_time,

328

token_week)))

329

# Define starting values:

330

# Value so far

331

value = datetime.timedelta()

332

found_token = None

333

# Following valid tokens

334

followers = frozenset((token_duration, ))

335

# String left to parse

336

s = duration

337

# Loop until end token is found

338

while found_token is not token_end:

339

# Search for any currently valid tokens

340

for token in followers:

341

match = token.regexp.match(s)

342

if match is not None:

343

# Token found

344

if token.value is not None:

345

# Value found, parse digits

346

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

347

# Add to value so far

348

value += factor * token.value

349

# Strip token from string

350

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

351

# Go to found token

352

found_token = token

353

# Set valid next tokens

354

followers = found_token.followers

355

break

356

else:

357

# No currently valid tokens were found

358

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

359

.format(duration))

360

# End token found

361

return value

362

363

364

def parse_pre_1_6_1_interval(interval):

365

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

366

and return a datetime.timedelta

367

368

>>> parse_pre_1_6_1_interval('7d')

369

datetime.timedelta(7)

370

>>> parse_pre_1_6_1_interval('60s')

371

datetime.timedelta(0, 60)

372

>>> parse_pre_1_6_1_interval('60m')

373

datetime.timedelta(0, 3600)

374

>>> parse_pre_1_6_1_interval('24h')

375

datetime.timedelta(1)

376

>>> parse_pre_1_6_1_interval('1w')

377

datetime.timedelta(7)

378

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

379

datetime.timedelta(0, 330)

380

>>> parse_pre_1_6_1_interval('')

381

datetime.timedelta(0)

382

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

383

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

384

datetime.timedelta(2, 480, 18000)

385

386

"""

387

388

value = datetime.timedelta(0)

389

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

390

391

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

392

if suffix == "d":

393

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

394

elif suffix == "s":

395

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

396

elif suffix == "m":

397

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

398

elif suffix == "h":

399

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

400

elif suffix == "w":

401

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

402

elif suffix == "":

403

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

404

return value

405

406

407

def check_option_syntax(parser, options):

408

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

409

argparse"""

410

411

def has_actions(options):

412

return any((options.enable,

413

options.disable,

414

options.bump_timeout,

415

options.start_checker,

416

options.stop_checker,

417

options.is_enabled,

418

options.remove,

419

options.checker is not None,

420

options.timeout is not None,

421

options.extended_timeout is not None,

422

options.interval is not None,

423

options.approved_by_default is not None,

424

options.approval_delay is not None,

425

options.approval_duration is not None,

426

options.host is not None,

427

options.secret is not None,

428

options.approve,

429

options.deny))

430

431

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

432

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

433

if options.verbose and has_actions(options):

434

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

435

if options.dump_json and (options.verbose

436

or has_actions(options)):

437

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

438

if options.all and not has_actions(options):

439

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

440

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

441

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

442

if options.remove:

443

options.remove = False

444

if has_actions(options) and not options.deny:

445

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

446

options.remove = True

447

448

449

def commands_from_options(options):

450

451

commands = []

452

453

if options.is_enabled:

454

commands.append(IsEnabledCmd())

455

456

if options.approve:

457

commands.append(ApproveCmd())

458

459

if options.deny:

460

commands.append(DenyCmd())

461

462

if options.remove:

463

commands.append(RemoveCmd())

464

465

if options.dump_json:

466

commands.append(DumpJSONCmd())

467

468

if options.enable:

469

commands.append(EnableCmd())

470

471

if options.disable:

472

commands.append(DisableCmd())

473

474

if options.bump_timeout:

475

commands.append(BumpTimeoutCmd())

476

477

if options.start_checker:

478

commands.append(StartCheckerCmd())

479

480

if options.stop_checker:

481

commands.append(StopCheckerCmd())

482

483

if options.approved_by_default is not None:

484

if options.approved_by_default:

485

commands.append(ApproveByDefaultCmd())

486

else:

487

commands.append(DenyByDefaultCmd())

488

489

if options.checker is not None:

490

commands.append(SetCheckerCmd(options.checker))

491

492

if options.host is not None:

493

commands.append(SetHostCmd(options.host))

494

495

if options.secret is not None:

496

commands.append(SetSecretCmd(options.secret))

497

498

if options.timeout is not None:

499

commands.append(SetTimeoutCmd(options.timeout))

500

501

if options.extended_timeout:

502

commands.append(

503

SetExtendedTimeoutCmd(options.extended_timeout))

504

505

if options.interval is not None:

506

commands.append(SetIntervalCmd(options.interval))

507

508

if options.approval_delay is not None:

509

commands.append(SetApprovalDelayCmd(options.approval_delay))

510

511

if options.approval_duration is not None:

512

commands.append(

513

SetApprovalDurationCmd(options.approval_duration))

514

515

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

516

# optionally verbosely

517

if not commands:

518

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

519

520

return commands

521

522

523

class Command(object):

524

"""Abstract class for commands"""

525

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

526

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

527

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

528

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

529

self.mandos = mandos

530

for clientpath, properties in clients.items():

531

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

532

dbus_busname, str(clientpath))

533

client = bus.get_object(dbus_busname, clientpath)

534

self.run_on_one_client(client, properties)

535

536

537

class IsEnabledCmd(Command):

538

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

539

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

540

if self.is_enabled(client, properties):

541

sys.exit(0)

542

sys.exit(1)

543

def is_enabled(self, client, properties):

544

return properties["Enabled"]

545

546

547

class ApproveCmd(Command):

548

def run_on_one_client(self, client, properties):

549

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

550

client.__dbus_object_path__, client_dbus_interface)

551

client.Approve(dbus.Boolean(True),

552

dbus_interface=client_dbus_interface)

553

554

555

class DenyCmd(Command):

556

def run_on_one_client(self, client, properties):

557

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

558

client.__dbus_object_path__, client_dbus_interface)

559

client.Approve(dbus.Boolean(False),

560

dbus_interface=client_dbus_interface)

561

562

563

class RemoveCmd(Command):

564

def run_on_one_client(self, client, properties):

565

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

566

server_dbus_path, server_dbus_interface,

567

str(client.__dbus_object_path__))

568

self.mandos.RemoveClient(client.__dbus_object_path__)

569

570

571

class OutputCmd(Command):

572

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

573

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

574

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

575

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

576

"ApprovalPending", "ApprovedByDefault",

577

"LastApprovalRequest", "ApprovalDelay",

578

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

579

"Expires", "LastCheckerStatus")

580

581

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

582

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

583

584

def output(self, clients):

585

raise NotImplementedError()

586

587

588

class DumpJSONCmd(OutputCmd):

589

def output(self, clients):

590

data = {client["Name"]:

591

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

592

for key in self.all_keywords}

593

for client in clients}

594

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

595

596

@staticmethod

597

def dbus_boolean_to_bool(value):

598

if isinstance(value, dbus.Boolean):

599

value = bool(value)

600

return value

601

602

603

class PrintTableCmd(OutputCmd):

604

def __init__(self, verbose=False):

605

self.verbose = verbose

606

607

def output(self, clients):

608

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

609

"LastCheckedOK")

610

keywords = default_keywords

611

if self.verbose:

612

keywords = self.all_keywords

613

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

614

615

class TableOfClients(object):

616

tableheaders = {

617

"Name": "Name",

618

"Enabled": "Enabled",

619

"Timeout": "Timeout",

620

"LastCheckedOK": "Last Successful Check",

621

"LastApprovalRequest": "Last Approval Request",

622

"Created": "Created",

623

"Interval": "Interval",

624

"Host": "Host",

625

"Fingerprint": "Fingerprint",

626

"KeyID": "Key ID",

627

"CheckerRunning": "Check Is Running",

628

"LastEnabled": "Last Enabled",

629

"ApprovalPending": "Approval Is Pending",

630

"ApprovedByDefault": "Approved By Default",

631

"ApprovalDelay": "Approval Delay",

632

"ApprovalDuration": "Approval Duration",

633

"Checker": "Checker",

634

"ExtendedTimeout": "Extended Timeout",

635

"Expires": "Expires",

636

"LastCheckerStatus": "Last Checker Status",

637

}

638

639

def __init__(self, clients, keywords):

640

self.clients = clients

641

self.keywords = keywords

642

643

def __str__(self):

644

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

645

646

if sys.version_info.major == 2:

647

__unicode__ = __str__

648

def __str__(self):

649

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

650

651

def rows(self):

652

format_string = self.row_formatting_string()

653

rows = [self.header_line(format_string)]

654

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

655

for client in self.clients)

656

return rows

657

658

def row_formatting_string(self):

659

"Format string used to format table rows"

660

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

661

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

662

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

663

for client in self.clients)),

664

key=key)

665

for key in self.keywords)

666

667

def string_from_client(self, client, key):

668

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

669

670

@classmethod

671

def valuetostring(cls, value, keyword):

672

if isinstance(value, dbus.Boolean):

673

return "Yes" if value else "No"

674

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

675

"ApprovalDuration", "ExtendedTimeout"):

676

return cls.milliseconds_to_string(value)

677

return str(value)

678

679

def header_line(self, format_string):

680

return format_string.format(**self.tableheaders)

681

682

def client_line(self, client, format_string):

683

return format_string.format(

684

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

685

for key in self.keywords})

686

687

@staticmethod

688

def milliseconds_to_string(ms):

689

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

690

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

691

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

692

if td.days else "",

693

hours=td.seconds // 3600,

694

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

695

seconds=td.seconds % 60))

696

697

698

class PropertyCmd(Command):

699

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

700

701

def run_on_one_client(self, client, properties):

702

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

703

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

704

client.__dbus_object_path__,

705

dbus.PROPERTIES_IFACE, client_dbus_interface,

706

self.propname, self.value_to_set

707

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

708

else bool(self.value_to_set))

709

client.Set(client_dbus_interface, self.propname,

710

self.value_to_set,

711

dbus_interface=dbus.PROPERTIES_IFACE)

712

713

@property

714

def propname(self):

715

raise NotImplementedError()

716

717

718

class EnableCmd(PropertyCmd):

719

propname = "Enabled"

720

value_to_set = dbus.Boolean(True)

721

722

723

class DisableCmd(PropertyCmd):

724

propname = "Enabled"

725

value_to_set = dbus.Boolean(False)

726

727

728

class BumpTimeoutCmd(PropertyCmd):

729

propname = "LastCheckedOK"

730

value_to_set = ""

731

732

733

class StartCheckerCmd(PropertyCmd):

734

propname = "CheckerRunning"

735

value_to_set = dbus.Boolean(True)

736

737

738

class StopCheckerCmd(PropertyCmd):

739

propname = "CheckerRunning"

740

value_to_set = dbus.Boolean(False)

741

742

743

class ApproveByDefaultCmd(PropertyCmd):

744

propname = "ApprovedByDefault"

745

value_to_set = dbus.Boolean(True)

746

747

748

class DenyByDefaultCmd(PropertyCmd):

749

propname = "ApprovedByDefault"

750

value_to_set = dbus.Boolean(False)

751

752

753

class PropertyValueCmd(PropertyCmd):

754

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

755

def __init__(self, value):

756

self.value_to_set = value

757

758

759

class SetCheckerCmd(PropertyValueCmd):

760

propname = "Checker"

761

762

763

class SetHostCmd(PropertyValueCmd):

764

propname = "Host"

765

766

767

class SetSecretCmd(PropertyValueCmd):

768

propname = "Secret"

769

770

@property

771

def value_to_set(self):

772

return self._vts

773

774

@value_to_set.setter

775

def value_to_set(self, value):

776

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

777

self._vts = value.read()

778

value.close()

779

780

781

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

782

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

783

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

784

785

@property

786

def value_to_set(self):

787

return self._vts

788

789

@value_to_set.setter

790

def value_to_set(self, value):

791

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

792

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

793

794

795

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

796

propname = "Timeout"

797

798

799

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

800

propname = "ExtendedTimeout"

801

802

803

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

804

propname = "Interval"

805

806

807

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

808

propname = "ApprovalDelay"

809

810

811

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

812

propname = "ApprovalDuration"

813

814

815

816

class Test_string_to_delta(unittest.TestCase):

817

def test_handles_basic_rfc3339(self):

818

self.assertEqual(string_to_delta("PT0S"),

819

datetime.timedelta())

820

self.assertEqual(string_to_delta("P0D"),

821

datetime.timedelta())

822

self.assertEqual(string_to_delta("PT1S"),

823

datetime.timedelta(0, 1))

824

self.assertEqual(string_to_delta("PT2H"),

825

datetime.timedelta(0, 7200))

826

827

def test_falls_back_to_pre_1_6_1_with_warning(self):

828

# assertLogs only exists in Python 3.4

829

if hasattr(self, "assertLogs"):

830

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

831

value = string_to_delta("2h")

832

else:

833

class WarningFilter(logging.Filter):

834

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

835

def filter(self, record):

836

is_warning = record.levelno >= logging.WARNING

837

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

838

False)

839

return not is_warning

840

warning_filter = WarningFilter()

841

log.addFilter(warning_filter)

842

try:

843

value = string_to_delta("2h")

844

finally:

845

log.removeFilter(warning_filter)

846

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

847

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

848

849

850

class Test_check_option_syntax(unittest.TestCase):

851

def setUp(self):

852

self.parser = argparse.ArgumentParser()

853

add_command_line_options(self.parser)

854

855

def test_actions_requires_client_or_all(self):

856

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

857

options = self.parser.parse_args()

858

setattr(options, action, value)

859

with self.assertParseError():

860

self.check_option_syntax(options)

861

862

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

863

# check_option_syntax()

864

actions = {

865

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

866

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

867

# never used

868

"enable": True,

869

"disable": True,

870

"bump_timeout": True,

871

"start_checker": True,

872

"stop_checker": True,

873

"is_enabled": True,

874

"remove": True,

875

"checker": "x",

876

"timeout": datetime.timedelta(),

877

"extended_timeout": datetime.timedelta(),

878

"interval": datetime.timedelta(),

879

"approved_by_default": True,

880

"approval_delay": datetime.timedelta(),

881

"approval_duration": datetime.timedelta(),

882

"host": "x",

883

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

884

"approve": True,

885

"deny": True,

886

}

887

888

@contextlib.contextmanager

889

def assertParseError(self):

890

with self.assertRaises(SystemExit) as e:

891

with self.temporarily_suppress_stderr():

892

yield

893

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

894

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

895

# /argparse.html#exiting-methods

896

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

897

898

@staticmethod

899

@contextlib.contextmanager

900

def temporarily_suppress_stderr():

901

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

902

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

903

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

904

os.close(null)

905

try:

906

yield

907

finally:

908

# restore stderr

909

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

910

os.close(stderrcopy)

911

912

def check_option_syntax(self, options):

913

check_option_syntax(self.parser, options)

914

915

def test_actions_conflicts_with_verbose(self):

916

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

917

options = self.parser.parse_args()

918

setattr(options, action, value)

919

options.verbose = True

920

with self.assertParseError():

921

self.check_option_syntax(options)

922

923

def test_dump_json_conflicts_with_verbose(self):

924

options = self.parser.parse_args()

925

options.dump_json = True

926

options.verbose = True

927

with self.assertParseError():

928

self.check_option_syntax(options)

929

930

def test_dump_json_conflicts_with_action(self):

931

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

932

options = self.parser.parse_args()

933

setattr(options, action, value)

934

options.dump_json = True

935

with self.assertParseError():

936

self.check_option_syntax(options)

937

938

def test_all_can_not_be_alone(self):

939

options = self.parser.parse_args()

940

options.all = True

941

with self.assertParseError():

942

self.check_option_syntax(options)

943

944

def test_all_is_ok_with_any_action(self):

945

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

946

options = self.parser.parse_args()

947

setattr(options, action, value)

948

options.all = True

949

self.check_option_syntax(options)

950

951

def test_is_enabled_fails_without_client(self):

952

options = self.parser.parse_args()

953

options.is_enabled = True

954

with self.assertParseError():

955

self.check_option_syntax(options)

956

957

def test_is_enabled_works_with_one_client(self):

958

options = self.parser.parse_args()

959

options.is_enabled = True

960

options.client = ["foo"]

961

self.check_option_syntax(options)

962

963

def test_is_enabled_fails_with_two_clients(self):

964

options = self.parser.parse_args()

965

options.is_enabled = True

966

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

967

with self.assertParseError():

968

self.check_option_syntax(options)

969

970

def test_remove_can_only_be_combined_with_action_deny(self):

971

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

972

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

973

continue

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.all = True

977

options.remove = True

978

with self.assertParseError():

979

self.check_option_syntax(options)

980

981

982

class Test_commands_from_options(unittest.TestCase):

983

def setUp(self):

984

self.parser = argparse.ArgumentParser()

985

add_command_line_options(self.parser)

986

987

def test_is_enabled(self):

988

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

989

IsEnabledCmd)

990

991

def assert_command_from_args(self, args, command_cls,

992

**cmd_attrs):

993

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

994

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

995

options = self.parser.parse_args(args)

996

check_option_syntax(self.parser, options)

997

commands = commands_from_options(options)

998

self.assertEqual(len(commands), 1)

999

command = commands[0]

1000

self.assertIsInstance(command, command_cls)

1001

for key, value in cmd_attrs.items():

1002

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

1003

1004

def test_is_enabled_short(self):

1005

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

1006

1007

def test_approve(self):

1008

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

1009

ApproveCmd)

1010

1011

def test_approve_short(self):

1012

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

1013

1014

def test_deny(self):

1015

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

1016

1017

def test_deny_short(self):

1018

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

1019

1020

def test_remove(self):

1021

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

1022

RemoveCmd)

1023

1024

def test_deny_before_remove(self):

1025

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

1026

"foo"])

1027

check_option_syntax(self.parser, options)

1028

commands = commands_from_options(options)

1029

self.assertEqual(len(commands), 2)

1030

self.assertIsInstance(commands[0], DenyCmd)

1031

self.assertIsInstance(commands[1], RemoveCmd)

1032

1033

def test_deny_before_remove_reversed(self):

1034

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

1035

"--all"])

1036

check_option_syntax(self.parser, options)

1037

commands = commands_from_options(options)

1038

self.assertEqual(len(commands), 2)

1039

self.assertIsInstance(commands[0], DenyCmd)

1040

self.assertIsInstance(commands[1], RemoveCmd)

1041

1042

def test_remove_short(self):

1043

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

1044

1045

def test_dump_json(self):

1046

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

1047

1048

def test_enable(self):

1049

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

1050

1051

def test_enable_short(self):

1052

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

1053

1054

def test_disable(self):

1055

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

1056

DisableCmd)

1057

1058

def test_disable_short(self):

1059

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

1060

1061

def test_bump_timeout(self):

1062

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

1063

BumpTimeoutCmd)

1064

1065

def test_bump_timeout_short(self):

1066

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

1067

1068

def test_start_checker(self):

1069

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

1070

StartCheckerCmd)

1071

1072

def test_stop_checker(self):

1073

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

1074

StopCheckerCmd)

1075

1076

def test_approve_by_default(self):

1077

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

1078

ApproveByDefaultCmd)

1079

1080

def test_deny_by_default(self):

1081

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

1082

DenyByDefaultCmd)

1083

1084

def test_checker(self):

1085

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

1086

SetCheckerCmd, value_to_set=":")

1087

1088

def test_checker_empty(self):

1089

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

1090

SetCheckerCmd, value_to_set="")

1091

1092

def test_checker_short(self):

1093

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

1094

SetCheckerCmd, value_to_set=":")

1095

1096

def test_host(self):

1097

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

1098

"foo"], SetHostCmd,

1099

value_to_set="foo.example.org")

1100

1101

def test_host_short(self):

1102

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

1103

"foo"], SetHostCmd,

1104

value_to_set="foo.example.org")

1105

1106

def test_secret_devnull(self):

1107

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

1108

"foo"], SetSecretCmd,

1109

value_to_set=b"")

1110

1111

def test_secret_tempfile(self):

1112

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

1113

value = b"secret\0xyzzy\nbar"

1114

f.write(value)

1115

f.seek(0)

1116

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

1117

"foo"], SetSecretCmd,

1118

value_to_set=value)

1119

1120

def test_secret_devnull_short(self):

1121

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

1122

SetSecretCmd, value_to_set=b"")

1123

1124

def test_secret_tempfile_short(self):

1125

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

1126

value = b"secret\0xyzzy\nbar"

1127

f.write(value)

1128

f.seek(0)

1129

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

1130

SetSecretCmd,

1131

value_to_set=value)

1132

1133

def test_timeout(self):

1134

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

1135

SetTimeoutCmd,

1136

value_to_set=300000)

1137

1138

def test_timeout_short(self):

1139

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

1140

SetTimeoutCmd,

1141

value_to_set=300000)

1142

1143

def test_extended_timeout(self):

1144

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

1145

"foo"],

1146

SetExtendedTimeoutCmd,

1147

value_to_set=900000)

1148

1149

def test_interval(self):

1150

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

1151

SetIntervalCmd,

1152

value_to_set=120000)

1153

1154

def test_interval_short(self):

1155

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

1156

SetIntervalCmd,

1157

value_to_set=120000)

1158

1159

def test_approval_delay(self):

1160

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

1161

"foo"], SetApprovalDelayCmd,

1162

value_to_set=30000)

1163

1164

def test_approval_duration(self):

1165

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

1166

"foo"], SetApprovalDurationCmd,

1167

value_to_set=1000)

1168

1169

def test_print_table(self):

1170

self.assert_command_from_args([], PrintTableCmd,

1171

verbose=False)

1172

1173

def test_print_table_verbose(self):

1174

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

1175

verbose=True)

1176

1177

def test_print_table_verbose_short(self):

1178

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

1179

verbose=True)

1180

1181

1182

class TestCmd(unittest.TestCase):

1183

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

1184

1185

def setUp(self):

1186

testcase = self

1187

class MockClient(object):

1188

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

1189

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

1190

self.attributes = attributes

1191

self.attributes["Name"] = name

1192

self.calls = []

1193

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

1194

testcase.assertEqual(interface, client_dbus_interface)

1195

testcase.assertEqual(dbus_interface,

1196

dbus.PROPERTIES_IFACE)

1197

self.attributes[propname] = value

1198

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

1199

testcase.assertEqual(interface, client_dbus_interface)

1200

testcase.assertEqual(dbus_interface,

1201

dbus.PROPERTIES_IFACE)

1202

return self.attributes[propname]

1203

def Approve(self, approve, dbus_interface):

1204

testcase.assertEqual(dbus_interface,

1205

client_dbus_interface)

1206

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

1207

dbus_interface)))

1208

self.client = MockClient(

1209

"foo",

1210

KeyID=("92ed150794387c03ce684574b1139a65"

1211

"94a34f895daaaf09fd8ea90a27cddb12"),

1212

Secret=b"secret",

1213

Host="foo.example.org",

1214

Enabled=dbus.Boolean(True),

1215

Timeout=300000,

1216

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

1217

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

1218

Interval=120000,

1219

Fingerprint=("778827225BA7DE539C5A"

1220

"7CFA59CFF7CDBD9A5920"),

1221

CheckerRunning=dbus.Boolean(False),

1222

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

1223

ApprovalPending=dbus.Boolean(False),

1224

ApprovedByDefault=dbus.Boolean(True),

1225

LastApprovalRequest="",

1226

ApprovalDelay=0,

1227

ApprovalDuration=1000,

1228

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

1229

ExtendedTimeout=900000,

1230

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

1231

LastCheckerStatus=0)

1232

self.other_client = MockClient(

1233

"barbar",

1234

KeyID=("0558568eedd67d622f5c83b35a115f79"

1235

"6ab612cff5ad227247e46c2b020f441c"),

1236

Secret=b"secretbar",

1237

Host="192.0.2.3",

1238

Enabled=dbus.Boolean(True),

1239

Timeout=300000,

1240

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

1241

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

1242

Interval=120000,

1243

Fingerprint=("3E393AEAEFB84C7E89E2"

1244

"F547B3A107558FCA3A27"),

1245

CheckerRunning=dbus.Boolean(True),

1246

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

1247

ApprovalPending=dbus.Boolean(False),

1248

ApprovedByDefault=dbus.Boolean(False),

1249

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

1250

ApprovalDelay=30000,

1251

ApprovalDuration=93785000,

1252

Checker=":",

1253

ExtendedTimeout=900000,

1254

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

1255

LastCheckerStatus=-2)

1256

self.clients = collections.OrderedDict(

1257

[

1258

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

1259

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

1260

])

1261

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

1262

1263

@property

1264

def bus(self):

1265

class Bus(object):

1266

@staticmethod

1267

def get_object(client_bus_name, path):

1268

self.assertEqual(client_bus_name, dbus_busname)

1269

return {

1270

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

1271

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

1272

"/clients/foo": self.client,

1273

"/clients/barbar": self.other_client,

1274

}[path]

1275

return Bus()

1276

1277

1278

class TestIsEnabledCmd(TestCmd):

1279

def test_is_enabled(self):

1280

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

1281

properties)

1282

for client, properties

1283

in self.clients.items()))

1284

1285

def test_is_enabled_run_exits_successfully(self):

1286

with self.assertRaises(SystemExit) as e:

1287

IsEnabledCmd().run(self.one_client)

1288

if e.exception.code is not None:

1289

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

1290

else:

1291

self.assertIsNone(e.exception.code)

1292

1293

def test_is_enabled_run_exits_with_failure(self):

1294

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

1295

with self.assertRaises(SystemExit) as e:

1296

IsEnabledCmd().run(self.one_client)

1297

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

1298

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

1299

else:

1300

self.assertIsNotNone(e.exception.code)

1301

1302

1303

class TestApproveCmd(TestCmd):

1304

def test_approve(self):

1305

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

1306

for clientpath in self.clients:

1307

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

1308

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

1309

client.calls)

1310

1311

1312

class TestDenyCmd(TestCmd):

1313

def test_deny(self):

1314

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

1315

for clientpath in self.clients:

1316

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

1317

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

1318

client.calls)

1319

1320

1321

class TestRemoveCmd(TestCmd):

1322

def test_remove(self):

1323

class MockMandos(object):

1324

def __init__(self):

1325

self.calls = []

1326

def RemoveClient(self, dbus_path):

1327

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

1328

mandos = MockMandos()

1329

super(TestRemoveCmd, self).setUp()

1330

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

1331

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

1332

for clientpath in self.clients:

1333

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

1334

mandos.calls)

1335

1336

1337

class TestDumpJSONCmd(TestCmd):

1338

def setUp(self):

1339

self.expected_json = {

1340

"foo": {

1341

"Name": "foo",

1342

"KeyID": ("92ed150794387c03ce684574b1139a65"

1343

"94a34f895daaaf09fd8ea90a27cddb12"),

1344

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

1345

"Enabled": True,

1346

"Timeout": 300000,

1347

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

1348

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

1349

"Interval": 120000,

1350

"Fingerprint": ("778827225BA7DE539C5A"

1351

"7CFA59CFF7CDBD9A5920"),

1352

"CheckerRunning": False,

1353

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

1354

"ApprovalPending": False,

1355

"ApprovedByDefault": True,

1356

"LastApprovalRequest": "",

1357

"ApprovalDelay": 0,

1358

"ApprovalDuration": 1000,

1359

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

1360

"ExtendedTimeout": 900000,

1361

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

1362

"LastCheckerStatus": 0,

1363

},

1364

"barbar": {

1365

"Name": "barbar",

1366

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1367

"6ab612cff5ad227247e46c2b020f441c"),

1368

"Host": "192.0.2.3",

1369

"Enabled": True,

1370

"Timeout": 300000,

1371

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

1372

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

1373

"Interval": 120000,

1374

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1375

"F547B3A107558FCA3A27"),

1376

"CheckerRunning": True,

1377

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

1378

"ApprovalPending": False,

1379

"ApprovedByDefault": False,

1380

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

1381

"ApprovalDelay": 30000,

1382

"ApprovalDuration": 93785000,

1383

"Checker": ":",

1384

"ExtendedTimeout": 900000,

1385

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

1386

"LastCheckerStatus": -2,

1387

},

1388

}

1389

return super(TestDumpJSONCmd, self).setUp()

1390

1391

def test_normal(self):

1392

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

1393

json_data = json.loads(output)

1394

self.assertDictEqual(json_data, self.expected_json)

1395

1396

def test_one_client(self):

1397

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

1398

json_data = json.loads(output)

1399

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

1400

self.assertDictEqual(json_data, expected_json)

1401

1402

1403

class TestPrintTableCmd(TestCmd):

1404

def test_normal(self):

1405

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

1406

expected_output = "\n".join((

1407

"Name Enabled Timeout Last Successful Check",

1408

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

1409

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

1410

))

1411

self.assertEqual(output, expected_output)

1412

1413

def test_verbose(self):

1414

output = PrintTableCmd(verbose=True).output(

1415

self.clients.values())

1416

columns = (

1417

(

1418

"Name ",

1419

"foo ",

1420

"barbar ",

1421

),(

1422

"Enabled ",

1423

"Yes ",

1424

"Yes ",

1425

),(

1426

"Timeout ",

1427

"00:05:00 ",

1428

"00:05:00 ",

1429

),(

1430

"Last Successful Check ",

1431

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

1432

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

1433

),(

1434

"Created ",

1435

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

1436

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

1437

),(

1438

"Interval ",

1439

"00:02:00 ",

1440

"00:02:00 ",

1441

),(

1442

"Host ",

1443

"foo.example.org ",

1444

"192.0.2.3 ",

1445

),(

1446

("Key ID "

1447

" "),

1448

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1449

"ea90a27cddb12 "),

1450

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1451

"46c2b020f441c "),

1452

),(

1453

"Fingerprint ",

1454

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1455

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1456

),(

1457

"Check Is Running ",

1458

"No ",

1459

"Yes ",

1460

),(

1461

"Last Enabled ",

1462

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

1463

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

1464

),(

1465

"Approval Is Pending ",

1466

"No ",

1467

"No ",

1468

),(

1469

"Approved By Default ",

1470

"Yes ",

1471

"No ",

1472

),(

1473

"Last Approval Request ",

1474

" ",

1475

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

1476

),(

1477

"Approval Delay ",

1478

"00:00:00 ",

1479

"00:00:30 ",

1480

),(

1481

"Approval Duration ",

1482

"00:00:01 ",

1483

"1T02:03:05 ",

1484

),(

1485

"Checker ",

1486

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

1487

": ",

1488

),(

1489

"Extended Timeout ",

1490

"00:15:00 ",

1491

"00:15:00 ",

1492

),(

1493

"Expires ",

1494

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

1495

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

1496

),(

1497

"Last Checker Status",

1498

"0 ",

1499

"-2 ",

1500

)

1501

)

1502

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

1503

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

1504

for rows in columns)

1505

for line in range(num_lines))

1506

self.assertEqual(output, expected_output)

1507

1508

def test_one_client(self):

1509

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

1510

expected_output = "\n".join((

1511

"Name Enabled Timeout Last Successful Check",

1512

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

1513

))

1514

self.assertEqual(output, expected_output)

1515

1516

1517

class TestPropertyCmd(TestCmd):

1518

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

1519

def runTest(self):

1520

if not hasattr(self, "command"):

1521

return

1522

values_to_get = getattr(self, "values_to_get",

1523

self.values_to_set)

1524

for value_to_set, value_to_get in zip(self.values_to_set,

1525

values_to_get):

1526

for clientpath in self.clients:

1527

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

1528

old_value = client.attributes[self.propname]

1529

self.assertNotIsInstance(old_value, self.Unique)

1530

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

1531

self.run_command(value_to_set, self.clients)

1532

for clientpath in self.clients:

1533

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

1534

value = client.attributes[self.propname]

1535

self.assertNotIsInstance(value, self.Unique)

1536

self.assertEqual(value, value_to_get)

1537

1538

class Unique(object):

1539

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

1540

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

1541

1542

def run_command(self, value, clients):

1543

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

1544

1545

1546

class TestEnableCmd(TestPropertyCmd):

1547

command = EnableCmd

1548

propname = "Enabled"

1549

values_to_set = [dbus.Boolean(True)]

1550

1551

1552

class TestDisableCmd(TestPropertyCmd):

1553

command = DisableCmd

1554

propname = "Enabled"

1555

values_to_set = [dbus.Boolean(False)]

1556

1557

1558

class TestBumpTimeoutCmd(TestPropertyCmd):

1559

command = BumpTimeoutCmd

1560

propname = "LastCheckedOK"

1561

values_to_set = [""]

1562

1563

1564

class TestStartCheckerCmd(TestPropertyCmd):

1565

command = StartCheckerCmd

1566

propname = "CheckerRunning"

1567

values_to_set = [dbus.Boolean(True)]

1568

1569

1570

class TestStopCheckerCmd(TestPropertyCmd):

1571

command = StopCheckerCmd

1572

propname = "CheckerRunning"

1573

values_to_set = [dbus.Boolean(False)]

1574

1575

1576

class TestApproveByDefaultCmd(TestPropertyCmd):

1577

command = ApproveByDefaultCmd

1578

propname = "ApprovedByDefault"

1579

values_to_set = [dbus.Boolean(True)]

1580

1581

1582

class TestDenyByDefaultCmd(TestPropertyCmd):

1583

command = DenyByDefaultCmd

1584

propname = "ApprovedByDefault"

1585

values_to_set = [dbus.Boolean(False)]

1586

1587

1588

class TestPropertyValueCmd(TestPropertyCmd):

1589

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

1590

1591

def runTest(self):

1592

if type(self) is TestPropertyValueCmd:

1593

return

1594

return super(TestPropertyValueCmd, self).runTest()

1595

1596

def run_command(self, value, clients):

1597

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

1598

1599

1600

class TestSetCheckerCmd(TestPropertyValueCmd):

1601

command = SetCheckerCmd

1602

propname = "Checker"

1603

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

1604

1605

1606

class TestSetHostCmd(TestPropertyValueCmd):

1607

command = SetHostCmd

1608

propname = "Host"

1609

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

1610

1611

1612

class TestSetSecretCmd(TestPropertyValueCmd):

1613

command = SetSecretCmd

1614

propname = "Secret"

1615

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

1616

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

1617

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

1618

1619

1620

class TestSetTimeoutCmd(TestPropertyValueCmd):

1621

command = SetTimeoutCmd

1622

propname = "Timeout"

1623

values_to_set = [datetime.timedelta(),

1624

datetime.timedelta(minutes=5),

1625

datetime.timedelta(seconds=1),

1626

datetime.timedelta(weeks=1),

1627

datetime.timedelta(weeks=52)]

1628

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

1629

1630

1631

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1632

command = SetExtendedTimeoutCmd

1633

propname = "ExtendedTimeout"

1634

values_to_set = [datetime.timedelta(),

1635

datetime.timedelta(minutes=5),

1636

datetime.timedelta(seconds=1),

1637

datetime.timedelta(weeks=1),

1638

datetime.timedelta(weeks=52)]

1639

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

1640

1641

1642

class TestSetIntervalCmd(TestPropertyValueCmd):

1643

command = SetIntervalCmd

1644

propname = "Interval"

1645

values_to_set = [datetime.timedelta(),

1646

datetime.timedelta(minutes=5),

1647

datetime.timedelta(seconds=1),

1648

datetime.timedelta(weeks=1),

1649

datetime.timedelta(weeks=52)]

1650

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

1651

1652

1653

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1654

command = SetApprovalDelayCmd

1655

propname = "ApprovalDelay"

1656

values_to_set = [datetime.timedelta(),

1657

datetime.timedelta(minutes=5),

1658

datetime.timedelta(seconds=1),

1659

datetime.timedelta(weeks=1),

1660

datetime.timedelta(weeks=52)]

1661

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

1662

1663

1664

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1665

command = SetApprovalDurationCmd

1666

propname = "ApprovalDuration"

1667

values_to_set = [datetime.timedelta(),

1668

datetime.timedelta(minutes=5),

1669

datetime.timedelta(seconds=1),

1670

datetime.timedelta(weeks=1),

1671

datetime.timedelta(weeks=52)]

1672

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

1673

1674

1675

1676

def should_only_run_tests():

1677

parser = argparse.ArgumentParser(add_help=False)

1678

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

1679

args, unknown_args = parser.parse_known_args()

1680

run_tests = args.check

1681

if run_tests:

1682

# Remove --check argument from sys.argv

1683

sys.argv[1:] = unknown_args

1684

return run_tests

1685

1686

# Add all tests from doctest strings

1687

def load_tests(loader, tests, none):

1688

import doctest

1689

tests.addTests(doctest.DocTestSuite())

1690

return tests

1691

1692

if __name__ == "__main__":

1693

if should_only_run_tests():

1694

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

1695

unittest.main()

1696

else:

1697

main()