/mandos/release : revision 174

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, tableheaders=None):

640

self.clients = clients

641

self.keywords = keywords

642

if tableheaders is not None:

643

self.tableheaders = tableheaders

644

645

def __str__(self):

646

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

647

648

if sys.version_info.major == 2:

649

__unicode__ = __str__

650

def __str__(self):

651

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

652

653

def rows(self):

654

format_string = self.row_formatting_string()

655

rows = [self.header_line(format_string)]

656

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

657

for client in self.clients)

658

return rows

659

660

def row_formatting_string(self):

661

"Format string used to format table rows"

662

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

663

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

664

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

665

for client in self.clients)),

666

key=key)

667

for key in self.keywords)

668

669

def string_from_client(self, client, key):

670

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

671

672

@classmethod

673

def valuetostring(cls, value, keyword):

674

if isinstance(value, dbus.Boolean):

675

return "Yes" if value else "No"

676

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

677

"ApprovalDuration", "ExtendedTimeout"):

678

return cls.milliseconds_to_string(value)

679

return str(value)

680

681

def header_line(self, format_string):

682

return format_string.format(**self.tableheaders)

683

684

def client_line(self, client, format_string):

685

return format_string.format(

686

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

687

for key in self.keywords})

688

689

@staticmethod

690

def milliseconds_to_string(ms):

691

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

692

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

693

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

694

if td.days else "",

695

hours=td.seconds // 3600,

696

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

697

seconds=td.seconds % 60))

698

699

700

class PropertyCmd(Command):

701

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

702

703

def run_on_one_client(self, client, properties):

704

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

705

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

706

client.__dbus_object_path__,

707

dbus.PROPERTIES_IFACE, client_dbus_interface,

708

self.propname, self.value_to_set

709

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

710

else bool(self.value_to_set))

711

client.Set(client_dbus_interface, self.propname,

712

self.value_to_set,

713

dbus_interface=dbus.PROPERTIES_IFACE)

714

715

@property

716

def propname(self):

717

raise NotImplementedError()

718

719

720

class EnableCmd(PropertyCmd):

721

propname = "Enabled"

722

value_to_set = dbus.Boolean(True)

723

724

725

class DisableCmd(PropertyCmd):

726

propname = "Enabled"

727

value_to_set = dbus.Boolean(False)

728

729

730

class BumpTimeoutCmd(PropertyCmd):

731

propname = "LastCheckedOK"

732

value_to_set = ""

733

734

735

class StartCheckerCmd(PropertyCmd):

736

propname = "CheckerRunning"

737

value_to_set = dbus.Boolean(True)

738

739

740

class StopCheckerCmd(PropertyCmd):

741

propname = "CheckerRunning"

742

value_to_set = dbus.Boolean(False)

743

744

745

class ApproveByDefaultCmd(PropertyCmd):

746

propname = "ApprovedByDefault"

747

value_to_set = dbus.Boolean(True)

748

749

750

class DenyByDefaultCmd(PropertyCmd):

751

propname = "ApprovedByDefault"

752

value_to_set = dbus.Boolean(False)

753

754

755

class PropertyValueCmd(PropertyCmd):

756

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

757

def __init__(self, value):

758

self.value_to_set = value

759

760

761

class SetCheckerCmd(PropertyValueCmd):

762

propname = "Checker"

763

764

765

class SetHostCmd(PropertyValueCmd):

766

propname = "Host"

767

768

769

class SetSecretCmd(PropertyValueCmd):

770

propname = "Secret"

771

772

@property

773

def value_to_set(self):

774

return self._vts

775

776

@value_to_set.setter

777

def value_to_set(self, value):

778

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

779

self._vts = value.read()

780

value.close()

781

782

783

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

784

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

785

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

786

787

@property

788

def value_to_set(self):

789

return self._vts

790

791

@value_to_set.setter

792

def value_to_set(self, value):

793

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

794

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

795

796

797

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

798

propname = "Timeout"

799

800

801

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

802

propname = "ExtendedTimeout"

803

804

805

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

806

propname = "Interval"

807

808

809

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

810

propname = "ApprovalDelay"

811

812

813

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

814

propname = "ApprovalDuration"

815

816

817

818

class Test_string_to_delta(unittest.TestCase):

819

def test_handles_basic_rfc3339(self):

820

self.assertEqual(string_to_delta("PT0S"),

821

datetime.timedelta())

822

self.assertEqual(string_to_delta("P0D"),

823

datetime.timedelta())

824

self.assertEqual(string_to_delta("PT1S"),

825

datetime.timedelta(0, 1))

826

self.assertEqual(string_to_delta("PT2H"),

827

datetime.timedelta(0, 7200))

828

829

def test_falls_back_to_pre_1_6_1_with_warning(self):

830

# assertLogs only exists in Python 3.4

831

if hasattr(self, "assertLogs"):

832

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

833

value = string_to_delta("2h")

834

else:

835

class WarningFilter(logging.Filter):

836

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

837

def filter(self, record):

838

is_warning = record.levelno >= logging.WARNING

839

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

840

False)

841

return not is_warning

842

warning_filter = WarningFilter()

843

log.addFilter(warning_filter)

844

try:

845

value = string_to_delta("2h")

846

finally:

847

log.removeFilter(warning_filter)

848

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

849

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

850

851

852

class Test_check_option_syntax(unittest.TestCase):

853

def setUp(self):

854

self.parser = argparse.ArgumentParser()

855

add_command_line_options(self.parser)

856

857

def test_actions_requires_client_or_all(self):

858

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

859

options = self.parser.parse_args()

860

setattr(options, action, value)

861

with self.assertParseError():

862

self.check_option_syntax(options)

863

864

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

865

# check_option_syntax()

866

actions = {

867

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

868

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

869

# never used

870

"enable": True,

871

"disable": True,

872

"bump_timeout": True,

873

"start_checker": True,

874

"stop_checker": True,

875

"is_enabled": True,

876

"remove": True,

877

"checker": "x",

878

"timeout": datetime.timedelta(),

879

"extended_timeout": datetime.timedelta(),

880

"interval": datetime.timedelta(),

881

"approved_by_default": True,

882

"approval_delay": datetime.timedelta(),

883

"approval_duration": datetime.timedelta(),

884

"host": "x",

885

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

886

"approve": True,

887

"deny": True,

888

}

889

890

@contextlib.contextmanager

891

def assertParseError(self):

892

with self.assertRaises(SystemExit) as e:

893

with self.temporarily_suppress_stderr():

894

yield

895

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

896

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

897

# /argparse.html#exiting-methods

898

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

899

900

@staticmethod

901

@contextlib.contextmanager

902

def temporarily_suppress_stderr():

903

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

904

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

905

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

906

os.close(null)

907

try:

908

yield

909

finally:

910

# restore stderr

911

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

912

os.close(stderrcopy)

913

914

def check_option_syntax(self, options):

915

check_option_syntax(self.parser, options)

916

917

def test_actions_conflicts_with_verbose(self):

918

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

919

options = self.parser.parse_args()

920

setattr(options, action, value)

921

options.verbose = True

922

with self.assertParseError():

923

self.check_option_syntax(options)

924

925

def test_dump_json_conflicts_with_verbose(self):

926

options = self.parser.parse_args()

927

options.dump_json = True

928

options.verbose = True

929

with self.assertParseError():

930

self.check_option_syntax(options)

931

932

def test_dump_json_conflicts_with_action(self):

933

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

934

options = self.parser.parse_args()

935

setattr(options, action, value)

936

options.dump_json = True

937

with self.assertParseError():

938

self.check_option_syntax(options)

939

940

def test_all_can_not_be_alone(self):

941

options = self.parser.parse_args()

942

options.all = True

943

with self.assertParseError():

944

self.check_option_syntax(options)

945

946

def test_all_is_ok_with_any_action(self):

947

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

948

options = self.parser.parse_args()

949

setattr(options, action, value)

950

options.all = True

951

self.check_option_syntax(options)

952

953

def test_is_enabled_fails_without_client(self):

954

options = self.parser.parse_args()

955

options.is_enabled = True

956

with self.assertParseError():

957

self.check_option_syntax(options)

958

959

def test_is_enabled_works_with_one_client(self):

960

options = self.parser.parse_args()

961

options.is_enabled = True

962

options.client = ["foo"]

963

self.check_option_syntax(options)

964

965

def test_is_enabled_fails_with_two_clients(self):

966

options = self.parser.parse_args()

967

options.is_enabled = True

968

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

969

with self.assertParseError():

970

self.check_option_syntax(options)

971

972

def test_remove_can_only_be_combined_with_action_deny(self):

973

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

974

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

975

continue

976

options = self.parser.parse_args()

977

setattr(options, action, value)

978

options.all = True

979

options.remove = True

980

with self.assertParseError():

981

self.check_option_syntax(options)

982

983

984

class Test_commands_from_options(unittest.TestCase):

985

def setUp(self):

986

self.parser = argparse.ArgumentParser()

987

add_command_line_options(self.parser)

988

989

def test_is_enabled(self):

990

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

991

IsEnabledCmd)

992

993

def assert_command_from_args(self, args, command_cls,

994

**cmd_attrs):

995

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

996

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

997

options = self.parser.parse_args(args)

998

check_option_syntax(self.parser, options)

999

commands = commands_from_options(options)

1000

self.assertEqual(len(commands), 1)

1001

command = commands[0]

1002

self.assertIsInstance(command, command_cls)

1003

for key, value in cmd_attrs.items():

1004

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

1005

1006

def test_is_enabled_short(self):

1007

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

1008

1009

def test_approve(self):

1010

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

1011

ApproveCmd)

1012

1013

def test_approve_short(self):

1014

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

1015

1016

def test_deny(self):

1017

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

1018

1019

def test_deny_short(self):

1020

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

1021

1022

def test_remove(self):

1023

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

1024

RemoveCmd)

1025

1026

def test_deny_before_remove(self):

1027

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

1028

"foo"])

1029

check_option_syntax(self.parser, options)

1030

commands = commands_from_options(options)

1031

self.assertEqual(len(commands), 2)

1032

self.assertIsInstance(commands[0], DenyCmd)

1033

self.assertIsInstance(commands[1], RemoveCmd)

1034

1035

def test_deny_before_remove_reversed(self):

1036

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

1037

"--all"])

1038

check_option_syntax(self.parser, options)

1039

commands = commands_from_options(options)

1040

self.assertEqual(len(commands), 2)

1041

self.assertIsInstance(commands[0], DenyCmd)

1042

self.assertIsInstance(commands[1], RemoveCmd)

1043

1044

def test_remove_short(self):

1045

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

1046

1047

def test_dump_json(self):

1048

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

1049

1050

def test_enable(self):

1051

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

1052

1053

def test_enable_short(self):

1054

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

1055

1056

def test_disable(self):

1057

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

1058

DisableCmd)

1059

1060

def test_disable_short(self):

1061

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

1062

1063

def test_bump_timeout(self):

1064

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

1065

BumpTimeoutCmd)

1066

1067

def test_bump_timeout_short(self):

1068

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

1069

1070

def test_start_checker(self):

1071

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

1072

StartCheckerCmd)

1073

1074

def test_stop_checker(self):

1075

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

1076

StopCheckerCmd)

1077

1078

def test_approve_by_default(self):

1079

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

1080

ApproveByDefaultCmd)

1081

1082

def test_deny_by_default(self):

1083

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

1084

DenyByDefaultCmd)

1085

1086

def test_checker(self):

1087

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

1088

SetCheckerCmd, value_to_set=":")

1089

1090

def test_checker_empty(self):

1091

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

1092

SetCheckerCmd, value_to_set="")

1093

1094

def test_checker_short(self):

1095

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

1096

SetCheckerCmd, value_to_set=":")

1097

1098

def test_host(self):

1099

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

1100

"foo"], SetHostCmd,

1101

value_to_set="foo.example.org")

1102

1103

def test_host_short(self):

1104

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

1105

"foo"], SetHostCmd,

1106

value_to_set="foo.example.org")

1107

1108

def test_secret_devnull(self):

1109

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

1110

"foo"], SetSecretCmd,

1111

value_to_set=b"")

1112

1113

def test_secret_tempfile(self):

1114

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

1115

value = b"secret\0xyzzy\nbar"

1116

f.write(value)

1117

f.seek(0)

1118

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

1119

"foo"], SetSecretCmd,

1120

value_to_set=value)

1121

1122

def test_secret_devnull_short(self):

1123

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

1124

SetSecretCmd, value_to_set=b"")

1125

1126

def test_secret_tempfile_short(self):

1127

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

1128

value = b"secret\0xyzzy\nbar"

1129

f.write(value)

1130

f.seek(0)

1131

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

1132

SetSecretCmd,

1133

value_to_set=value)

1134

1135

def test_timeout(self):

1136

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

1137

SetTimeoutCmd,

1138

value_to_set=300000)

1139

1140

def test_timeout_short(self):

1141

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

1142

SetTimeoutCmd,

1143

value_to_set=300000)

1144

1145

def test_extended_timeout(self):

1146

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

1147

"foo"],

1148

SetExtendedTimeoutCmd,

1149

value_to_set=900000)

1150

1151

def test_interval(self):

1152

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

1153

SetIntervalCmd,

1154

value_to_set=120000)

1155

1156

def test_interval_short(self):

1157

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

1158

SetIntervalCmd,

1159

value_to_set=120000)

1160

1161

def test_approval_delay(self):

1162

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

1163

"foo"], SetApprovalDelayCmd,

1164

value_to_set=30000)

1165

1166

def test_approval_duration(self):

1167

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

1168

"foo"], SetApprovalDurationCmd,

1169

value_to_set=1000)

1170

1171

def test_print_table(self):

1172

self.assert_command_from_args([], PrintTableCmd,

1173

verbose=False)

1174

1175

def test_print_table_verbose(self):

1176

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

1177

verbose=True)

1178

1179

def test_print_table_verbose_short(self):

1180

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

1181

verbose=True)

1182

1183

1184

class TestCmd(unittest.TestCase):

1185

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

1186

1187

def setUp(self):

1188

testcase = self

1189

class MockClient(object):

1190

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

1191

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

1192

self.attributes = attributes

1193

self.attributes["Name"] = name

1194

self.calls = []

1195

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

1196

testcase.assertEqual(interface, client_dbus_interface)

1197

testcase.assertEqual(dbus_interface,

1198

dbus.PROPERTIES_IFACE)

1199

self.attributes[propname] = value

1200

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

1201

testcase.assertEqual(interface, client_dbus_interface)

1202

testcase.assertEqual(dbus_interface,

1203

dbus.PROPERTIES_IFACE)

1204

return self.attributes[propname]

1205

def Approve(self, approve, dbus_interface):

1206

testcase.assertEqual(dbus_interface,

1207

client_dbus_interface)

1208

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

1209

dbus_interface)))

1210

self.client = MockClient(

1211

"foo",

1212

KeyID=("92ed150794387c03ce684574b1139a65"

1213

"94a34f895daaaf09fd8ea90a27cddb12"),

1214

Secret=b"secret",

1215

Host="foo.example.org",

1216

Enabled=dbus.Boolean(True),

1217

Timeout=300000,

1218

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

1219

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

1220

Interval=120000,

1221

Fingerprint=("778827225BA7DE539C5A"

1222

"7CFA59CFF7CDBD9A5920"),

1223

CheckerRunning=dbus.Boolean(False),

1224

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

1225

ApprovalPending=dbus.Boolean(False),

1226

ApprovedByDefault=dbus.Boolean(True),

1227

LastApprovalRequest="",

1228

ApprovalDelay=0,

1229

ApprovalDuration=1000,

1230

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

1231

ExtendedTimeout=900000,

1232

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

1233

LastCheckerStatus=0)

1234

self.other_client = MockClient(

1235

"barbar",

1236

KeyID=("0558568eedd67d622f5c83b35a115f79"

1237

"6ab612cff5ad227247e46c2b020f441c"),

1238

Secret=b"secretbar",

1239

Host="192.0.2.3",

1240

Enabled=dbus.Boolean(True),

1241

Timeout=300000,

1242

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

1243

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

1244

Interval=120000,

1245

Fingerprint=("3E393AEAEFB84C7E89E2"

1246

"F547B3A107558FCA3A27"),

1247

CheckerRunning=dbus.Boolean(True),

1248

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

1249

ApprovalPending=dbus.Boolean(False),

1250

ApprovedByDefault=dbus.Boolean(False),

1251

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

1252

ApprovalDelay=30000,

1253

ApprovalDuration=93785000,

1254

Checker=":",

1255

ExtendedTimeout=900000,

1256

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

1257

LastCheckerStatus=-2)

1258

self.clients = collections.OrderedDict(

1259

[

1260

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

1261

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

1262

])

1263

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

1264

1265

@property

1266

def bus(self):

1267

class Bus(object):

1268

@staticmethod

1269

def get_object(client_bus_name, path):

1270

self.assertEqual(client_bus_name, dbus_busname)

1271

return {

1272

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

1273

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

1274

"/clients/foo": self.client,

1275

"/clients/barbar": self.other_client,

1276

}[path]

1277

return Bus()

1278

1279

1280

class TestIsEnabledCmd(TestCmd):

1281

def test_is_enabled(self):

1282

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

1283

properties)

1284

for client, properties

1285

in self.clients.items()))

1286

1287

def test_is_enabled_run_exits_successfully(self):

1288

with self.assertRaises(SystemExit) as e:

1289

IsEnabledCmd().run(self.one_client)

1290

if e.exception.code is not None:

1291

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

1292

else:

1293

self.assertIsNone(e.exception.code)

1294

1295

def test_is_enabled_run_exits_with_failure(self):

1296

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

1297

with self.assertRaises(SystemExit) as e:

1298

IsEnabledCmd().run(self.one_client)

1299

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

1300

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

1301

else:

1302

self.assertIsNotNone(e.exception.code)

1303

1304

1305

class TestApproveCmd(TestCmd):

1306

def test_approve(self):

1307

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

1308

for clientpath in self.clients:

1309

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

1310

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

1311

client.calls)

1312

1313

1314

class TestDenyCmd(TestCmd):

1315

def test_deny(self):

1316

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

1317

for clientpath in self.clients:

1318

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

1319

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

1320

client.calls)

1321

1322

1323

class TestRemoveCmd(TestCmd):

1324

def test_remove(self):

1325

class MockMandos(object):

1326

def __init__(self):

1327

self.calls = []

1328

def RemoveClient(self, dbus_path):

1329

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

1330

mandos = MockMandos()

1331

super(TestRemoveCmd, self).setUp()

1332

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

1333

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

1334

for clientpath in self.clients:

1335

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

1336

mandos.calls)

1337

1338

1339

class TestDumpJSONCmd(TestCmd):

1340

def setUp(self):

1341

self.expected_json = {

1342

"foo": {

1343

"Name": "foo",

1344

"KeyID": ("92ed150794387c03ce684574b1139a65"

1345

"94a34f895daaaf09fd8ea90a27cddb12"),

1346

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

1347

"Enabled": True,

1348

"Timeout": 300000,

1349

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

1350

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

1351

"Interval": 120000,

1352

"Fingerprint": ("778827225BA7DE539C5A"

1353

"7CFA59CFF7CDBD9A5920"),

1354

"CheckerRunning": False,

1355

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

1356

"ApprovalPending": False,

1357

"ApprovedByDefault": True,

1358

"LastApprovalRequest": "",

1359

"ApprovalDelay": 0,

1360

"ApprovalDuration": 1000,

1361

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

1362

"ExtendedTimeout": 900000,

1363

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

1364

"LastCheckerStatus": 0,

1365

},

1366

"barbar": {

1367

"Name": "barbar",

1368

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1369

"6ab612cff5ad227247e46c2b020f441c"),

1370

"Host": "192.0.2.3",

1371

"Enabled": True,

1372

"Timeout": 300000,

1373

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

1374

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

1375

"Interval": 120000,

1376

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1377

"F547B3A107558FCA3A27"),

1378

"CheckerRunning": True,

1379

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

1380

"ApprovalPending": False,

1381

"ApprovedByDefault": False,

1382

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

1383

"ApprovalDelay": 30000,

1384

"ApprovalDuration": 93785000,

1385

"Checker": ":",

1386

"ExtendedTimeout": 900000,

1387

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

1388

"LastCheckerStatus": -2,

1389

},

1390

}

1391

return super(TestDumpJSONCmd, self).setUp()

1392

1393

def test_normal(self):

1394

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

1395

json_data = json.loads(output)

1396

self.assertDictEqual(json_data, self.expected_json)

1397

1398

def test_one_client(self):

1399

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

1400

json_data = json.loads(output)

1401

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

1402

self.assertDictEqual(json_data, expected_json)

1403

1404

1405

class TestPrintTableCmd(TestCmd):

1406

def test_normal(self):

1407

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

1408

expected_output = "\n".join((

1409

"Name Enabled Timeout Last Successful Check",

1410

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

1411

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

1412

))

1413

self.assertEqual(output, expected_output)

1414

1415

def test_verbose(self):

1416

output = PrintTableCmd(verbose=True).output(

1417

self.clients.values())

1418

columns = (

1419

(

1420

"Name ",

1421

"foo ",

1422

"barbar ",

1423

),(

1424

"Enabled ",

1425

"Yes ",

1426

"Yes ",

1427

),(

1428

"Timeout ",

1429

"00:05:00 ",

1430

"00:05:00 ",

1431

),(

1432

"Last Successful Check ",

1433

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

1434

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

1435

),(

1436

"Created ",

1437

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

1438

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

1439

),(

1440

"Interval ",

1441

"00:02:00 ",

1442

"00:02:00 ",

1443

),(

1444

"Host ",

1445

"foo.example.org ",

1446

"192.0.2.3 ",

1447

),(

1448

("Key ID "

1449

" "),

1450

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1451

"ea90a27cddb12 "),

1452

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1453

"46c2b020f441c "),

1454

),(

1455

"Fingerprint ",

1456

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1457

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1458

),(

1459

"Check Is Running ",

1460

"No ",

1461

"Yes ",

1462

),(

1463

"Last Enabled ",

1464

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

1465

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

1466

),(

1467

"Approval Is Pending ",

1468

"No ",

1469

"No ",

1470

),(

1471

"Approved By Default ",

1472

"Yes ",

1473

"No ",

1474

),(

1475

"Last Approval Request ",

1476

" ",

1477

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

1478

),(

1479

"Approval Delay ",

1480

"00:00:00 ",

1481

"00:00:30 ",

1482

),(

1483

"Approval Duration ",

1484

"00:00:01 ",

1485

"1T02:03:05 ",

1486

),(

1487

"Checker ",

1488

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

1489

": ",

1490

),(

1491

"Extended Timeout ",

1492

"00:15:00 ",

1493

"00:15:00 ",

1494

),(

1495

"Expires ",

1496

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

1497

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

1498

),(

1499

"Last Checker Status",

1500

"0 ",

1501

"-2 ",

1502

)

1503

)

1504

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

1505

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

1506

for rows in columns)

1507

for line in range(num_lines))

1508

self.assertEqual(output, expected_output)

1509

1510

def test_one_client(self):

1511

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

1512

expected_output = "\n".join((

1513

"Name Enabled Timeout Last Successful Check",

1514

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

1515

))

1516

self.assertEqual(output, expected_output)

1517

1518

1519

class TestPropertyCmd(TestCmd):

1520

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

1521

def runTest(self):

1522

if not hasattr(self, "command"):

1523

return

1524

values_to_get = getattr(self, "values_to_get",

1525

self.values_to_set)

1526

for value_to_set, value_to_get in zip(self.values_to_set,

1527

values_to_get):

1528

for clientpath in self.clients:

1529

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

1530

old_value = client.attributes[self.propname]

1531

self.assertNotIsInstance(old_value, self.Unique)

1532

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

1533

self.run_command(value_to_set, self.clients)

1534

for clientpath in self.clients:

1535

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

1536

value = client.attributes[self.propname]

1537

self.assertNotIsInstance(value, self.Unique)

1538

self.assertEqual(value, value_to_get)

1539

1540

class Unique(object):

1541

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

1542

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

1543

1544

def run_command(self, value, clients):

1545

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

1546

1547

1548

class TestEnableCmd(TestPropertyCmd):

1549

command = EnableCmd

1550

propname = "Enabled"

1551

values_to_set = [dbus.Boolean(True)]

1552

1553

1554

class TestDisableCmd(TestPropertyCmd):

1555

command = DisableCmd

1556

propname = "Enabled"

1557

values_to_set = [dbus.Boolean(False)]

1558

1559

1560

class TestBumpTimeoutCmd(TestPropertyCmd):

1561

command = BumpTimeoutCmd

1562

propname = "LastCheckedOK"

1563

values_to_set = [""]

1564

1565

1566

class TestStartCheckerCmd(TestPropertyCmd):

1567

command = StartCheckerCmd

1568

propname = "CheckerRunning"

1569

values_to_set = [dbus.Boolean(True)]

1570

1571

1572

class TestStopCheckerCmd(TestPropertyCmd):

1573

command = StopCheckerCmd

1574

propname = "CheckerRunning"

1575

values_to_set = [dbus.Boolean(False)]

1576

1577

1578

class TestApproveByDefaultCmd(TestPropertyCmd):

1579

command = ApproveByDefaultCmd

1580

propname = "ApprovedByDefault"

1581

values_to_set = [dbus.Boolean(True)]

1582

1583

1584

class TestDenyByDefaultCmd(TestPropertyCmd):

1585

command = DenyByDefaultCmd

1586

propname = "ApprovedByDefault"

1587

values_to_set = [dbus.Boolean(False)]

1588

1589

1590

class TestPropertyValueCmd(TestPropertyCmd):

1591

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

1592

1593

def runTest(self):

1594

if type(self) is TestPropertyValueCmd:

1595

return

1596

return super(TestPropertyValueCmd, self).runTest()

1597

1598

def run_command(self, value, clients):

1599

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

1600

1601

1602

class TestSetCheckerCmd(TestPropertyValueCmd):

1603

command = SetCheckerCmd

1604

propname = "Checker"

1605

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

1606

1607

1608

class TestSetHostCmd(TestPropertyValueCmd):

1609

command = SetHostCmd

1610

propname = "Host"

1611

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

1612

1613

1614

class TestSetSecretCmd(TestPropertyValueCmd):

1615

command = SetSecretCmd

1616

propname = "Secret"

1617

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

1618

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

1619

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

1620

1621

1622

class TestSetTimeoutCmd(TestPropertyValueCmd):

1623

command = SetTimeoutCmd

1624

propname = "Timeout"

1625

values_to_set = [datetime.timedelta(),

1626

datetime.timedelta(minutes=5),

1627

datetime.timedelta(seconds=1),

1628

datetime.timedelta(weeks=1),

1629

datetime.timedelta(weeks=52)]

1630

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

1631

1632

1633

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1634

command = SetExtendedTimeoutCmd

1635

propname = "ExtendedTimeout"

1636

values_to_set = [datetime.timedelta(),

1637

datetime.timedelta(minutes=5),

1638

datetime.timedelta(seconds=1),

1639

datetime.timedelta(weeks=1),

1640

datetime.timedelta(weeks=52)]

1641

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

1642

1643

1644

class TestSetIntervalCmd(TestPropertyValueCmd):

1645

command = SetIntervalCmd

1646

propname = "Interval"

1647

values_to_set = [datetime.timedelta(),

1648

datetime.timedelta(minutes=5),

1649

datetime.timedelta(seconds=1),

1650

datetime.timedelta(weeks=1),

1651

datetime.timedelta(weeks=52)]

1652

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

1653

1654

1655

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1656

command = SetApprovalDelayCmd

1657

propname = "ApprovalDelay"

1658

values_to_set = [datetime.timedelta(),

1659

datetime.timedelta(minutes=5),

1660

datetime.timedelta(seconds=1),

1661

datetime.timedelta(weeks=1),

1662

datetime.timedelta(weeks=52)]

1663

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

1664

1665

1666

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1667

command = SetApprovalDurationCmd

1668

propname = "ApprovalDuration"

1669

values_to_set = [datetime.timedelta(),

1670

datetime.timedelta(minutes=5),

1671

datetime.timedelta(seconds=1),

1672

datetime.timedelta(weeks=1),

1673

datetime.timedelta(weeks=52)]

1674

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

1675

1676

1677

1678

def should_only_run_tests():

1679

parser = argparse.ArgumentParser(add_help=False)

1680

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

1681

args, unknown_args = parser.parse_known_args()

1682

run_tests = args.check

1683

if run_tests:

1684

# Remove --check argument from sys.argv

1685

sys.argv[1:] = unknown_args

1686

return run_tests

1687

1688

# Add all tests from doctest strings

1689

def load_tests(loader, tests, none):

1690

import doctest

1691

tests.addTests(doctest.DocTestSuite())

1692

return tests

1693

1694

if __name__ == "__main__":

1695

if should_only_run_tests():

1696

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

1697

unittest.main()

1698

else:

1699

main()