/mandos/trunk : revision 179

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

domain = "se.recompile"

68

busname = domain + ".Mandos"

69

server_path = "/"

70

server_interface = domain + ".Mandos"

71

client_interface = domain + ".Mandos.Client"

72

version = "1.8.3"

73

74

75

try:

76

dbus.OBJECT_MANAGER_IFACE

77

except AttributeError:

78

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

79

80

81

def milliseconds_to_string(ms):

82

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

83

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

84

.format(days="{}T".format(td.days) if td.days else "",

85

hours=td.seconds // 3600,

86

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

87

seconds=td.seconds % 60))

88

89

90

def rfc3339_duration_to_delta(duration):

91

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

92

93

>>> rfc3339_duration_to_delta("P7D")

94

datetime.timedelta(7)

95

>>> rfc3339_duration_to_delta("PT60S")

96

datetime.timedelta(0, 60)

97

>>> rfc3339_duration_to_delta("PT60M")

98

datetime.timedelta(0, 3600)

99

>>> rfc3339_duration_to_delta("P60M")

100

datetime.timedelta(1680)

101

>>> rfc3339_duration_to_delta("PT24H")

102

datetime.timedelta(1)

103

>>> rfc3339_duration_to_delta("P1W")

104

datetime.timedelta(7)

105

>>> rfc3339_duration_to_delta("PT5M30S")

106

datetime.timedelta(0, 330)

107

>>> rfc3339_duration_to_delta("P1DT3M20S")

108

datetime.timedelta(1, 200)

109

>>> # Can not be empty:

110

>>> rfc3339_duration_to_delta("")

111

Traceback (most recent call last):

112

...

113

ValueError: Invalid RFC 3339 duration: u''

114

>>> # Must start with "P":

115

>>> rfc3339_duration_to_delta("1D")

116

Traceback (most recent call last):

117

...

118

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

119

>>> # Must use correct order

120

>>> rfc3339_duration_to_delta("PT1S2M")

121

Traceback (most recent call last):

122

...

123

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

124

>>> # Time needs time marker

125

>>> rfc3339_duration_to_delta("P1H2S")

126

Traceback (most recent call last):

127

...

128

ValueError: Invalid RFC 3339 duration: u'P1H2S'

129

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

130

>>> rfc3339_duration_to_delta("P1D2W")

131

Traceback (most recent call last):

132

...

133

ValueError: Invalid RFC 3339 duration: u'P1D2W'

134

>>> rfc3339_duration_to_delta("P2W2H")

135

Traceback (most recent call last):

136

...

137

ValueError: Invalid RFC 3339 duration: u'P2W2H'

138

"""

139

140

# Parsing an RFC 3339 duration with regular expressions is not

141

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

142

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

143

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

144

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

145

# avoid excessive use of external libraries.

146

147

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

148

Token = collections.namedtuple("Token", (

149

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

150

# a "group" containing digits

151

"value", # datetime.timedelta or None

152

"followers")) # Tokens valid after this token

153

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

154

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

155

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

156

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

157

datetime.timedelta(seconds=1),

158

frozenset((token_end, )))

159

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

160

datetime.timedelta(minutes=1),

161

frozenset((token_second, token_end)))

162

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

163

datetime.timedelta(hours=1),

164

frozenset((token_minute, token_end)))

165

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

166

None,

167

frozenset((token_hour, token_minute,

168

token_second)))

169

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

170

datetime.timedelta(days=1),

171

frozenset((token_time, token_end)))

172

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

173

datetime.timedelta(weeks=4),

174

frozenset((token_day, token_end)))

175

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

176

datetime.timedelta(weeks=52),

177

frozenset((token_month, token_end)))

178

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

179

datetime.timedelta(weeks=1),

180

frozenset((token_end, )))

181

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

182

frozenset((token_year, token_month,

183

token_day, token_time,

184

token_week)))

185

# Define starting values:

186

# Value so far

187

value = datetime.timedelta()

188

found_token = None

189

# Following valid tokens

190

followers = frozenset((token_duration, ))

191

# String left to parse

192

s = duration

193

# Loop until end token is found

194

while found_token is not token_end:

195

# Search for any currently valid tokens

196

for token in followers:

197

match = token.regexp.match(s)

198

if match is not None:

199

# Token found

200

if token.value is not None:

201

# Value found, parse digits

202

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

203

# Add to value so far

204

value += factor * token.value

205

# Strip token from string

206

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

207

# Go to found token

208

found_token = token

209

# Set valid next tokens

210

followers = found_token.followers

211

break

212

else:

213

# No currently valid tokens were found

214

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

215

.format(duration))

216

# End token found

217

return value

218

219

220

def string_to_delta(interval):

221

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

222

223

try:

224

return rfc3339_duration_to_delta(interval)

225

except ValueError as e:

226

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

227

' '.join(e.args))

228

return parse_pre_1_6_1_interval(interval)

229

230

231

def parse_pre_1_6_1_interval(interval):

232

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

233

and return a datetime.timedelta

234

235

>>> parse_pre_1_6_1_interval('7d')

236

datetime.timedelta(7)

237

>>> parse_pre_1_6_1_interval('60s')

238

datetime.timedelta(0, 60)

239

>>> parse_pre_1_6_1_interval('60m')

240

datetime.timedelta(0, 3600)

241

>>> parse_pre_1_6_1_interval('24h')

242

datetime.timedelta(1)

243

>>> parse_pre_1_6_1_interval('1w')

244

datetime.timedelta(7)

245

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

246

datetime.timedelta(0, 330)

247

>>> parse_pre_1_6_1_interval('')

248

datetime.timedelta(0)

249

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

250

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

251

datetime.timedelta(2, 480, 18000)

252

253

"""

254

255

value = datetime.timedelta(0)

256

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

257

258

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

259

if suffix == "d":

260

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

261

elif suffix == "s":

262

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

263

elif suffix == "m":

264

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

265

elif suffix == "h":

266

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

267

elif suffix == "w":

268

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

269

elif suffix == "":

270

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

271

return value

272

273

274

## Classes for commands.

275

276

# Abstract classes first

277

class Command(object):

278

"""Abstract class for commands"""

279

def run(self, mandos, clients):

280

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

281

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

282

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

283

self.mandos = mandos

284

for client, properties in clients.items():

285

self.run_on_one_client(client, properties)

286

287

class PrintCmd(Command):

288

"""Abstract class for commands printing client details"""

289

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

290

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

291

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

292

"ApprovalPending", "ApprovedByDefault",

293

"LastApprovalRequest", "ApprovalDelay",

294

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

295

"Expires", "LastCheckerStatus")

296

def run(self, mandos, clients):

297

print(self.output(clients))

298

299

class PropertyCmd(Command):

300

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

301

def run_on_one_client(self, client, properties):

302

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

303

client.Set(client_interface, self.property, self.value_to_set,

304

dbus_interface=dbus.PROPERTIES_IFACE)

305

306

class ValueArgumentMixIn(object):

307

"""Mixin class for commands taking a value as argument"""

308

def __init__(self, value):

309

self.value_to_set = value

310

311

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

312

"""Mixin class for commands taking a value argument as

313

milliseconds."""

314

@property

315

def value_to_set(self):

316

return self._vts

317

@value_to_set.setter

318

def value_to_set(self, value):

319

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

320

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

321

322

# Actual (non-abstract) command classes

323

324

class PrintTableCmd(PrintCmd):

325

def __init__(self, verbose=False):

326

self.verbose = verbose

327

328

def output(self, clients):

329

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

330

keywords = default_keywords

331

if self.verbose:

332

keywords = self.all_keywords

333

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

334

335

class TableOfClients(object):

336

tableheaders = {

337

"Name": "Name",

338

"Enabled": "Enabled",

339

"Timeout": "Timeout",

340

"LastCheckedOK": "Last Successful Check",

341

"LastApprovalRequest": "Last Approval Request",

342

"Created": "Created",

343

"Interval": "Interval",

344

"Host": "Host",

345

"Fingerprint": "Fingerprint",

346

"KeyID": "Key ID",

347

"CheckerRunning": "Check Is Running",

348

"LastEnabled": "Last Enabled",

349

"ApprovalPending": "Approval Is Pending",

350

"ApprovedByDefault": "Approved By Default",

351

"ApprovalDelay": "Approval Delay",

352

"ApprovalDuration": "Approval Duration",

353

"Checker": "Checker",

354

"ExtendedTimeout": "Extended Timeout",

355

"Expires": "Expires",

356

"LastCheckerStatus": "Last Checker Status",

357

}

358

359

def __init__(self, clients, keywords, tableheaders=None):

360

self.clients = clients

361

self.keywords = keywords

362

if tableheaders is not None:

363

self.tableheaders = tableheaders

364

365

def __str__(self):

366

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

367

368

if sys.version_info.major == 2:

369

__unicode__ = __str__

370

def __str__(self):

371

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

372

373

def rows(self):

374

format_string = self.row_formatting_string()

375

rows = [self.header_line(format_string)]

376

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

377

for client in self.clients)

378

return rows

379

380

def row_formatting_string(self):

381

"Format string used to format table rows"

382

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

383

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

384

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

385

for client in self.clients)),

386

key=key)

387

for key in self.keywords)

388

389

def string_from_client(self, client, key):

390

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

391

392

@staticmethod

393

def valuetostring(value, keyword):

394

if isinstance(value, dbus.Boolean):

395

return "Yes" if value else "No"

396

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

397

"ApprovalDuration", "ExtendedTimeout"):

398

return milliseconds_to_string(value)

399

return str(value)

400

401

def header_line(self, format_string):

402

return format_string.format(**self.tableheaders)

403

404

def client_line(self, client, format_string):

405

return format_string.format(

406

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

407

for key in self.keywords})

408

409

410

411

class DumpJSONCmd(PrintCmd):

412

def output(self, clients):

413

data = {client["Name"]:

414

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

415

for key in self.all_keywords}

416

for client in clients.values()}

417

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

418

@staticmethod

419

def dbus_boolean_to_bool(value):

420

if isinstance(value, dbus.Boolean):

421

value = bool(value)

422

return value

423

424

class IsEnabledCmd(Command):

425

def run_on_one_client(self, client, properties):

426

if self.is_enabled(client, properties):

427

sys.exit(0)

428

sys.exit(1)

429

def is_enabled(self, client, properties):

430

return bool(properties["Enabled"])

431

432

class RemoveCmd(Command):

433

def run_on_one_client(self, client, properties):

434

self.mandos.RemoveClient(client.__dbus_object_path__)

435

436

class ApproveCmd(Command):

437

def run_on_one_client(self, client, properties):

438

client.Approve(dbus.Boolean(True),

439

dbus_interface=client_interface)

440

441

class DenyCmd(Command):

442

def run_on_one_client(self, client, properties):

443

client.Approve(dbus.Boolean(False),

444

dbus_interface=client_interface)

445

446

class EnableCmd(PropertyCmd):

447

property = "Enabled"

448

value_to_set = dbus.Boolean(True)

449

450

class DisableCmd(PropertyCmd):

451

property = "Enabled"

452

value_to_set = dbus.Boolean(False)

453

454

class BumpTimeoutCmd(PropertyCmd):

455

property = "LastCheckedOK"

456

value_to_set = ""

457

458

class StartCheckerCmd(PropertyCmd):

459

property = "CheckerRunning"

460

value_to_set = dbus.Boolean(True)

461

462

class StopCheckerCmd(PropertyCmd):

463

property = "CheckerRunning"

464

value_to_set = dbus.Boolean(False)

465

466

class ApproveByDefaultCmd(PropertyCmd):

467

property = "ApprovedByDefault"

468

value_to_set = dbus.Boolean(True)

469

470

class DenyByDefaultCmd(PropertyCmd):

471

property = "ApprovedByDefault"

472

value_to_set = dbus.Boolean(False)

473

474

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

475

property = "Checker"

476

477

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

478

property = "Host"

479

480

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

481

@property

482

def value_to_set(self):

483

return self._vts

484

@value_to_set.setter

485

def value_to_set(self, value):

486

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

487

self._vts = value.read()

488

value.close()

489

property = "Secret"

490

491

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

492

property = "Timeout"

493

494

class SetExtendedTimeoutCmd(PropertyCmd,

495

MillisecondsValueArgumentMixIn):

496

property = "ExtendedTimeout"

497

498

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

499

property = "Interval"

500

501

class SetApprovalDelayCmd(PropertyCmd,

502

MillisecondsValueArgumentMixIn):

503

property = "ApprovalDelay"

504

505

class SetApprovalDurationCmd(PropertyCmd,

506

MillisecondsValueArgumentMixIn):

507

property = "ApprovalDuration"

508

509

def add_command_line_options(parser):

510

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

511

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

512

help="show version number and exit")

513

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

514

help="Select all clients")

515

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

516

help="Print all fields")

517

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

518

help="Dump client data in JSON format")

519

enable_disable = parser.add_mutually_exclusive_group()

520

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

521

help="Enable client")

522

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

523

action="store_true",

524

help="disable client")

525

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

526

help="Bump timeout for client")

527

start_stop_checker = parser.add_mutually_exclusive_group()

528

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

529

action="store_true",

530

help="Start checker for client")

531

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

532

action="store_true",

533

help="Stop checker for client")

534

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

535

help="Check if client is enabled")

536

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

537

help="Remove client")

538

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

539

help="Set checker command for client")

540

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

541

help="Set timeout for client")

542

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

543

help="Set extended timeout for client")

544

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

545

help="Set checker interval for client")

546

approve_deny_default = parser.add_mutually_exclusive_group()

547

approve_deny_default.add_argument(

548

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

549

default=None, dest="approved_by_default",

550

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

551

approve_deny_default.add_argument(

552

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

553

dest="approved_by_default",

554

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

555

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

556

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

557

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

558

help="Set duration of one client approval")

559

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

560

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

561

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

562

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

563

approve_deny = parser.add_mutually_exclusive_group()

564

approve_deny.add_argument(

565

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

566

help="Approve any current client request")

567

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

568

help="Deny any current client request")

569

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

570

help="Run self-test")

571

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

572

573

574

def commands_from_options(options):

575

576

commands = []

577

578

if options.dump_json:

579

commands.append(DumpJSONCmd())

580

581

if options.enable:

582

commands.append(EnableCmd())

583

584

if options.disable:

585

commands.append(DisableCmd())

586

587

if options.bump_timeout:

588

commands.append(BumpTimeoutCmd())

589

590

if options.start_checker:

591

commands.append(StartCheckerCmd())

592

593

if options.stop_checker:

594

commands.append(StopCheckerCmd())

595

596

if options.is_enabled:

597

commands.append(IsEnabledCmd())

598

599

if options.remove:

600

commands.append(RemoveCmd())

601

602

if options.checker is not None:

603

commands.append(SetCheckerCmd(options.checker))

604

605

if options.timeout is not None:

606

commands.append(SetTimeoutCmd(options.timeout))

607

608

if options.extended_timeout:

609

commands.append(

610

SetExtendedTimeoutCmd(options.extended_timeout))

611

612

if options.interval is not None:

613

commands.append(SetIntervalCmd(options.interval))

614

615

if options.approved_by_default is not None:

616

if options.approved_by_default:

617

commands.append(ApproveByDefaultCmd())

618

else:

619

commands.append(DenyByDefaultCmd())

620

621

if options.approval_delay is not None:

622

commands.append(SetApprovalDelayCmd(options.approval_delay))

623

624

if options.approval_duration is not None:

625

commands.append(

626

SetApprovalDurationCmd(options.approval_duration))

627

628

if options.host is not None:

629

commands.append(SetHostCmd(options.host))

630

631

if options.secret is not None:

632

commands.append(SetSecretCmd(options.secret))

633

634

if options.approve:

635

commands.append(ApproveCmd())

636

637

if options.deny:

638

commands.append(DenyCmd())

639

640

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

641

# optionally verbosely

642

if not commands:

643

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

644

645

return commands

646

647

648

def check_option_syntax(parser, options):

649

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

650

argparse"""

651

652

def has_actions(options):

653

return any((options.enable,

654

options.disable,

655

options.bump_timeout,

656

options.start_checker,

657

options.stop_checker,

658

options.is_enabled,

659

options.remove,

660

options.checker is not None,

661

options.timeout is not None,

662

options.extended_timeout is not None,

663

options.interval is not None,

664

options.approved_by_default is not None,

665

options.approval_delay is not None,

666

options.approval_duration is not None,

667

options.host is not None,

668

options.secret is not None,

669

options.approve,

670

options.deny))

671

672

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

673

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

674

if options.verbose and has_actions(options):

675

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

676

if options.dump_json and (options.verbose

677

or has_actions(options)):

678

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

679

if options.all and not has_actions(options):

680

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

681

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

682

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

683

684

685

def main():

686

parser = argparse.ArgumentParser()

687

688

add_command_line_options(parser)

689

690

options = parser.parse_args()

691

692

check_option_syntax(parser, options)

693

694

clientnames = options.client

695

696

try:

697

bus = dbus.SystemBus()

698

mandos_dbus_objc = bus.get_object(busname, server_path)

699

except dbus.exceptions.DBusException:

700

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

701

sys.exit(1)

702

703

mandos_serv = dbus.Interface(mandos_dbus_objc,

704

dbus_interface=server_interface)

705

mandos_serv_object_manager = dbus.Interface(

706

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

707

708

# Filter out log message from dbus module

709

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

710

class NullFilter(logging.Filter):

711

def filter(self, record):

712

return False

713

dbus_filter = NullFilter()

714

try:

715

dbus_logger.addFilter(dbus_filter)

716

mandos_clients = {path: ifs_and_props[client_interface]

717

for path, ifs_and_props in

718

mandos_serv_object_manager

719

.GetManagedObjects().items()

720

if client_interface in ifs_and_props}

721

except dbus.exceptions.DBusException as e:

722

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

723

"\n%s", e)

724

sys.exit(1)

725

finally:

726

# restore dbus logger

727

dbus_logger.removeFilter(dbus_filter)

728

729

# Compile dict of (clients: properties) to process

730

clients = {}

731

732

if not clientnames:

733

clients = {bus.get_object(busname, path): properties

734

for path, properties in mandos_clients.items()}

735

else:

736

for name in clientnames:

737

for path, client in mandos_clients.items():

738

if client["Name"] == name:

739

client_objc = bus.get_object(busname, path)

740

clients[client_objc] = client

741

break

742

else:

743

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

744

sys.exit(1)

745

746

# Run all commands on clients

747

commands = commands_from_options(options)

748

for command in commands:

749

command.run(mandos_serv, clients)

750

751

752

class Test_milliseconds_to_string(unittest.TestCase):

753

def test_all(self):

754

self.assertEqual(milliseconds_to_string(93785000),

755

"1T02:03:05")

756

def test_no_days(self):

757

self.assertEqual(milliseconds_to_string(7385000), "02:03:05")

758

def test_all_zero(self):

759

self.assertEqual(milliseconds_to_string(0), "00:00:00")

760

def test_no_fractional_seconds(self):

761

self.assertEqual(milliseconds_to_string(400), "00:00:00")

762

self.assertEqual(milliseconds_to_string(900), "00:00:00")

763

self.assertEqual(milliseconds_to_string(1900), "00:00:01")

764

765

class Test_string_to_delta(unittest.TestCase):

766

def test_handles_basic_rfc3339(self):

767

self.assertEqual(string_to_delta("PT0S"),

768

datetime.timedelta())

769

self.assertEqual(string_to_delta("P0D"),

770

datetime.timedelta())

771

self.assertEqual(string_to_delta("PT1S"),

772

datetime.timedelta(0, 1))

773

self.assertEqual(string_to_delta("PT2H"),

774

datetime.timedelta(0, 7200))

775

def test_falls_back_to_pre_1_6_1_with_warning(self):

776

# assertLogs only exists in Python 3.4

777

if hasattr(self, "assertLogs"):

778

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

779

value = string_to_delta("2h")

780

else:

781

class WarningFilter(logging.Filter):

782

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

783

def filter(self, record):

784

is_warning = record.levelno >= logging.WARNING

785

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

786

False)

787

return not is_warning

788

warning_filter = WarningFilter()

789

log.addFilter(warning_filter)

790

try:

791

value = string_to_delta("2h")

792

finally:

793

log.removeFilter(warning_filter)

794

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

795

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

796

797

798

class TestCmd(unittest.TestCase):

799

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

800

def setUp(self):

801

testcase = self

802

class MockClient(object):

803

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

804

self.__dbus_object_path__ = "objpath_{}".format(name)

805

self.attributes = attributes

806

self.attributes["Name"] = name

807

self.calls = []

808

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

809

testcase.assertEqual(interface, client_interface)

810

testcase.assertEqual(dbus_interface,

811

dbus.PROPERTIES_IFACE)

812

self.attributes[property] = value

813

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

814

testcase.assertEqual(interface, client_interface)

815

testcase.assertEqual(dbus_interface,

816

dbus.PROPERTIES_IFACE)

817

return self.attributes[property]

818

def Approve(self, approve, dbus_interface):

819

testcase.assertEqual(dbus_interface, client_interface)

820

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

821

dbus_interface)))

822

self.client = MockClient(

823

"foo",

824

KeyID=("92ed150794387c03ce684574b1139a65"

825

"94a34f895daaaf09fd8ea90a27cddb12"),

826

Secret=b"secret",

827

Host="foo.example.org",

828

Enabled=dbus.Boolean(True),

829

Timeout=300000,

830

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

831

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

832

Interval=120000,

833

Fingerprint=("778827225BA7DE539C5A"

834

"7CFA59CFF7CDBD9A5920"),

835

CheckerRunning=dbus.Boolean(False),

836

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

837

ApprovalPending=dbus.Boolean(False),

838

ApprovedByDefault=dbus.Boolean(True),

839

LastApprovalRequest="",

840

ApprovalDelay=0,

841

ApprovalDuration=1000,

842

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

843

ExtendedTimeout=900000,

844

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

845

LastCheckerStatus=0)

846

self.other_client = MockClient(

847

"barbar",

848

KeyID=("0558568eedd67d622f5c83b35a115f79"

849

"6ab612cff5ad227247e46c2b020f441c"),

850

Secret=b"secretbar",

851

Host="192.0.2.3",

852

Enabled=dbus.Boolean(True),

853

Timeout=300000,

854

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

855

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

856

Interval=120000,

857

Fingerprint=("3E393AEAEFB84C7E89E2"

858

"F547B3A107558FCA3A27"),

859

CheckerRunning=dbus.Boolean(True),

860

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

861

ApprovalPending=dbus.Boolean(False),

862

ApprovedByDefault=dbus.Boolean(False),

863

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

864

ApprovalDelay=30000,

865

ApprovalDuration=1000,

866

Checker=":",

867

ExtendedTimeout=900000,

868

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

869

LastCheckerStatus=-2)

870

self.clients = collections.OrderedDict(

871

[

872

(self.client, self.client.attributes),

873

(self.other_client, self.other_client.attributes),

874

])

875

self.one_client = {self.client: self.client.attributes}

876

877

class TestPrintTableCmd(TestCmd):

878

def test_normal(self):

879

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

880

expected_output = """

881

Name Enabled Timeout Last Successful Check

882

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

883

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

884

"""[1:-1]

885

self.assertEqual(output, expected_output)

886

def test_verbose(self):

887

output = PrintTableCmd(verbose=True).output(self.clients)

888

expected_output = """

889

Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status

890

foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0

891

barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2

892

"""[1:-1]

893

self.assertEqual(output, expected_output)

894

def test_one_client(self):

895

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

896

expected_output = """

897

Name Enabled Timeout Last Successful Check

898

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

899

"""[1:-1]

900

self.assertEqual(output, expected_output)

901

902

class TestDumpJSONCmd(TestCmd):

903

def setUp(self):

904

self.expected_json = {

905

"foo": {

906

"Name": "foo",

907

"KeyID": ("92ed150794387c03ce684574b1139a65"

908

"94a34f895daaaf09fd8ea90a27cddb12"),

909

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

910

"Enabled": True,

911

"Timeout": 300000,

912

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

913

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

914

"Interval": 120000,

915

"Fingerprint": ("778827225BA7DE539C5A"

916

"7CFA59CFF7CDBD9A5920"),

917

"CheckerRunning": False,

918

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

919

"ApprovalPending": False,

920

"ApprovedByDefault": True,

921

"LastApprovalRequest": "",

922

"ApprovalDelay": 0,

923

"ApprovalDuration": 1000,

924

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

925

"ExtendedTimeout": 900000,

926

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

927

"LastCheckerStatus": 0,

928

},

929

"barbar": {

930

"Name": "barbar",

931

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

932

"6ab612cff5ad227247e46c2b020f441c"),

933

"Host": "192.0.2.3",

934

"Enabled": True,

935

"Timeout": 300000,

936

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

937

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

938

"Interval": 120000,

939

"Fingerprint": ("3E393AEAEFB84C7E89E2"

940

"F547B3A107558FCA3A27"),

941

"CheckerRunning": True,

942

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

943

"ApprovalPending": False,

944

"ApprovedByDefault": False,

945

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

946

"ApprovalDelay": 30000,

947

"ApprovalDuration": 1000,

948

"Checker": ":",

949

"ExtendedTimeout": 900000,

950

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

951

"LastCheckerStatus": -2,

952

},

953

}

954

return super(TestDumpJSONCmd, self).setUp()

955

def test_normal(self):

956

json_data = json.loads(DumpJSONCmd().output(self.clients))

957

self.assertDictEqual(json_data, self.expected_json)

958

def test_one_client(self):

959

clients = self.one_client

960

json_data = json.loads(DumpJSONCmd().output(clients))

961

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

962

self.assertDictEqual(json_data, expected_json)

963

964

class TestIsEnabledCmd(TestCmd):

965

def test_is_enabled(self):

966

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

967

for client, properties in self.clients.items()))

968

def test_is_enabled_run_exits_successfully(self):

969

with self.assertRaises(SystemExit) as e:

970

IsEnabledCmd().run(None, self.one_client)

971

if e.exception.code is not None:

972

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

973

else:

974

self.assertIsNone(e.exception.code)

975

def test_is_enabled_run_exits_with_failure(self):

976

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

977

with self.assertRaises(SystemExit) as e:

978

IsEnabledCmd().run(None, self.one_client)

979

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

980

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

981

else:

982

self.assertIsNotNone(e.exception.code)

983

984

class TestRemoveCmd(TestCmd):

985

def test_remove(self):

986

class MockMandos(object):

987

def __init__(self):

988

self.calls = []

989

def RemoveClient(self, dbus_path):

990

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

991

mandos = MockMandos()

992

super(TestRemoveCmd, self).setUp()

993

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

994

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

995

for client in self.clients:

996

self.assertIn(("RemoveClient",

997

(client.__dbus_object_path__,)),

998

mandos.calls)

999

1000

class TestApproveCmd(TestCmd):

1001

def test_approve(self):

1002

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

1003

for client in self.clients:

1004

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

1005

client.calls)

1006

1007

class TestDenyCmd(TestCmd):

1008

def test_deny(self):

1009

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

1010

for client in self.clients:

1011

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

1012

client.calls)

1013

1014

class TestEnableCmd(TestCmd):

1015

def test_enable(self):

1016

for client in self.clients:

1017

client.attributes["Enabled"] = False

1018

1019

EnableCmd().run(None, self.clients)

1020

1021

for client in self.clients:

1022

self.assertTrue(client.attributes["Enabled"])

1023

1024

class TestDisableCmd(TestCmd):

1025

def test_disable(self):

1026

DisableCmd().run(None, self.clients)

1027

1028

for client in self.clients:

1029

self.assertFalse(client.attributes["Enabled"])

1030

1031

class Unique(object):

1032

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

1033

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

1034

1035

class TestPropertyCmd(TestCmd):

1036

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

1037

def runTest(self):

1038

if not hasattr(self, "command"):

1039

return

1040

values_to_get = getattr(self, "values_to_get",

1041

self.values_to_set)

1042

for value_to_set, value_to_get in zip(self.values_to_set,

1043

values_to_get):

1044

for client in self.clients:

1045

old_value = client.attributes[self.property]

1046

self.assertNotIsInstance(old_value, Unique)

1047

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

1048

self.run_command(value_to_set, self.clients)

1049

for client in self.clients:

1050

value = client.attributes[self.property]

1051

self.assertNotIsInstance(value, Unique)

1052

self.assertEqual(value, value_to_get)

1053

def run_command(self, value, clients):

1054

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

1055

1056

class TestBumpTimeoutCmd(TestPropertyCmd):

1057

command = BumpTimeoutCmd

1058

property = "LastCheckedOK"

1059

values_to_set = [""]

1060

1061

class TestStartCheckerCmd(TestPropertyCmd):

1062

command = StartCheckerCmd

1063

property = "CheckerRunning"

1064

values_to_set = [dbus.Boolean(True)]

1065

1066

class TestStopCheckerCmd(TestPropertyCmd):

1067

command = StopCheckerCmd

1068

property = "CheckerRunning"

1069

values_to_set = [dbus.Boolean(False)]

1070

1071

class TestApproveByDefaultCmd(TestPropertyCmd):

1072

command = ApproveByDefaultCmd

1073

property = "ApprovedByDefault"

1074

values_to_set = [dbus.Boolean(True)]

1075

1076

class TestDenyByDefaultCmd(TestPropertyCmd):

1077

command = DenyByDefaultCmd

1078

property = "ApprovedByDefault"

1079

values_to_set = [dbus.Boolean(False)]

1080

1081

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1082

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

1083

ValueArgumentMixIn"""

1084

def runTest(self):

1085

if type(self) is TestValueArgumentPropertyCmd:

1086

return

1087

return super(TestValueArgumentPropertyCmd, self).runTest()

1088

def run_command(self, value, clients):

1089

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

1090

1091

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1092

command = SetCheckerCmd

1093

property = "Checker"

1094

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

1095

1096

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1097

command = SetHostCmd

1098

property = "Host"

1099

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

1100

1101

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1102

command = SetSecretCmd

1103

property = "Secret"

1104

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

1105

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

1106

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

1107

1108

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1109

command = SetTimeoutCmd

1110

property = "Timeout"

1111

values_to_set = [datetime.timedelta(),

1112

datetime.timedelta(minutes=5),

1113

datetime.timedelta(seconds=1),

1114

datetime.timedelta(weeks=1),

1115

datetime.timedelta(weeks=52)]

1116

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

1117

1118

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1119

command = SetExtendedTimeoutCmd

1120

property = "ExtendedTimeout"

1121

values_to_set = [datetime.timedelta(),

1122

datetime.timedelta(minutes=5),

1123

datetime.timedelta(seconds=1),

1124

datetime.timedelta(weeks=1),

1125

datetime.timedelta(weeks=52)]

1126

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

1127

1128

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1129

command = SetIntervalCmd

1130

property = "Interval"

1131

values_to_set = [datetime.timedelta(),

1132

datetime.timedelta(minutes=5),

1133

datetime.timedelta(seconds=1),

1134

datetime.timedelta(weeks=1),

1135

datetime.timedelta(weeks=52)]

1136

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

1137

1138

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1139

command = SetApprovalDelayCmd

1140

property = "ApprovalDelay"

1141

values_to_set = [datetime.timedelta(),

1142

datetime.timedelta(minutes=5),

1143

datetime.timedelta(seconds=1),

1144

datetime.timedelta(weeks=1),

1145

datetime.timedelta(weeks=52)]

1146

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

1147

1148

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1149

command = SetApprovalDurationCmd

1150

property = "ApprovalDuration"

1151

values_to_set = [datetime.timedelta(),

1152

datetime.timedelta(minutes=5),

1153

datetime.timedelta(seconds=1),

1154

datetime.timedelta(weeks=1),

1155

datetime.timedelta(weeks=52)]

1156

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

1157

1158

class Test_command_from_options(unittest.TestCase):

1159

def setUp(self):

1160

self.parser = argparse.ArgumentParser()

1161

add_command_line_options(self.parser)

1162

def assert_command_from_args(self, args, command_cls, **cmd_attrs):

1163

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

1164

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

1165

options = self.parser.parse_args(args)

1166

check_option_syntax(self.parser, options)

1167

commands = commands_from_options(options)

1168

self.assertEqual(len(commands), 1)

1169

command = commands[0]

1170

self.assertIsInstance(command, command_cls)

1171

for key, value in cmd_attrs.items():

1172

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

1173

def test_print_table(self):

1174

self.assert_command_from_args([], PrintTableCmd,

1175

verbose=False)

1176

1177

def test_print_table_verbose(self):

1178

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

1179

verbose=True)

1180

1181

def test_print_table_verbose_short(self):

1182

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

1183

verbose=True)

1184

1185

def test_enable(self):

1186

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

1187

1188

def test_enable_short(self):

1189

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

1190

1191

def test_disable(self):

1192

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

1193

DisableCmd)

1194

1195

def test_disable_short(self):

1196

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

1197

1198

def test_bump_timeout(self):

1199

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

1200

BumpTimeoutCmd)

1201

1202

def test_bump_timeout_short(self):

1203

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

1204

1205

def test_start_checker(self):

1206

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

1207

StartCheckerCmd)

1208

1209

def test_stop_checker(self):

1210

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

1211

StopCheckerCmd)

1212

1213

def test_remove(self):

1214

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

1215

RemoveCmd)

1216

1217

def test_remove_short(self):

1218

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

1219

1220

def test_checker(self):

1221

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

1222

SetCheckerCmd, value_to_set=":")

1223

1224

def test_checker_empty(self):

1225

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

1226

SetCheckerCmd, value_to_set="")

1227

1228

def test_checker_short(self):

1229

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

1230

SetCheckerCmd, value_to_set=":")

1231

1232

def test_timeout(self):

1233

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

1234

SetTimeoutCmd,

1235

value_to_set=300000)

1236

1237

def test_timeout_short(self):

1238

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

1239

SetTimeoutCmd,

1240

value_to_set=300000)

1241

1242

def test_extended_timeout(self):

1243

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

1244

"foo"],

1245

SetExtendedTimeoutCmd,

1246

value_to_set=900000)

1247

1248

def test_interval(self):

1249

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

1250

SetIntervalCmd,

1251

value_to_set=120000)

1252

1253

def test_interval_short(self):

1254

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

1255

SetIntervalCmd,

1256

value_to_set=120000)

1257

1258

def test_approve_by_default(self):

1259

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

1260

ApproveByDefaultCmd)

1261

1262

def test_deny_by_default(self):

1263

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

1264

DenyByDefaultCmd)

1265

1266

def test_approval_delay(self):

1267

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

1268

"foo"], SetApprovalDelayCmd,

1269

value_to_set=30000)

1270

1271

def test_approval_duration(self):

1272

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

1273

"foo"], SetApprovalDurationCmd,

1274

value_to_set=1000)

1275

1276

def test_host(self):

1277

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

1278

"foo"], SetHostCmd,

1279

value_to_set="foo.example.org")

1280

1281

def test_host_short(self):

1282

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

1283

"foo"], SetHostCmd,

1284

value_to_set="foo.example.org")

1285

1286

def test_secret_devnull(self):

1287

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

1288

"foo"], SetSecretCmd,

1289

value_to_set=b"")

1290

1291

def test_secret_tempfile(self):

1292

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

1293

value = b"secret\0xyzzy\nbar"

1294

f.write(value)

1295

f.seek(0)

1296

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

1297

"foo"], SetSecretCmd,

1298

value_to_set=value)

1299

1300

def test_secret_devnull_short(self):

1301

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

1302

SetSecretCmd, value_to_set=b"")

1303

1304

def test_secret_tempfile_short(self):

1305

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

1306

value = b"secret\0xyzzy\nbar"

1307

f.write(value)

1308

f.seek(0)

1309

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

1310

SetSecretCmd,

1311

value_to_set=value)

1312

1313

def test_approve(self):

1314

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

1315

ApproveCmd)

1316

1317

def test_approve_short(self):

1318

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

1319

1320

def test_deny(self):

1321

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

1322

1323

def test_deny_short(self):

1324

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

1325

1326

def test_dump_json(self):

1327

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

1328

1329

def test_is_enabled(self):

1330

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

1331

IsEnabledCmd)

1332

1333

def test_is_enabled_short(self):

1334

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

1335

1336

1337

class Test_check_option_syntax(unittest.TestCase):

1338

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

1339

# check_option_syntax()

1340

actions = {

1341

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

1342

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

1343

# never used

1344

"enable": True,

1345

"disable": True,

1346

"bump_timeout": True,

1347

"start_checker": True,

1348

"stop_checker": True,

1349

"is_enabled": True,

1350

"remove": True,

1351

"checker": "x",

1352

"timeout": datetime.timedelta(),

1353

"extended_timeout": datetime.timedelta(),

1354

"interval": datetime.timedelta(),

1355

"approved_by_default": True,

1356

"approval_delay": datetime.timedelta(),

1357

"approval_duration": datetime.timedelta(),

1358

"host": "x",

1359

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

1360

"approve": True,

1361

"deny": True,

1362

}

1363

1364

def setUp(self):

1365

self.parser = argparse.ArgumentParser()

1366

add_command_line_options(self.parser)

1367

1368

@contextlib.contextmanager

1369

def assertParseError(self):

1370

with self.assertRaises(SystemExit) as e:

1371

with self.temporarily_suppress_stderr():

1372

yield

1373

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

1374

# https://docs.python.org/3/library/argparse.html#exiting-methods

1375

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

1376

1377

@staticmethod

1378

@contextlib.contextmanager

1379

def temporarily_suppress_stderr():

1380

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

1381

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

1382

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

1383

os.close(null)

1384

try:

1385

yield

1386

finally:

1387

# restore stderr

1388

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

1389

os.close(stderrcopy)

1390

1391

def check_option_syntax(self, options):

1392

check_option_syntax(self.parser, options)

1393

1394

def test_actions_requires_client_or_all(self):

1395

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

1396

options = self.parser.parse_args()

1397

setattr(options, action, value)

1398

with self.assertParseError():

1399

self.check_option_syntax(options)

1400

1401

def test_actions_conflicts_with_verbose(self):

1402

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

1403

options = self.parser.parse_args()

1404

setattr(options, action, value)

1405

options.verbose = True

1406

with self.assertParseError():

1407

self.check_option_syntax(options)

1408

1409

def test_dump_json_conflicts_with_verbose(self):

1410

options = self.parser.parse_args()

1411

options.dump_json = True

1412

options.verbose = True

1413

with self.assertParseError():

1414

self.check_option_syntax(options)

1415

1416

def test_dump_json_conflicts_with_action(self):

1417

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

1418

options = self.parser.parse_args()

1419

setattr(options, action, value)

1420

options.dump_json = True

1421

with self.assertParseError():

1422

self.check_option_syntax(options)

1423

1424

def test_all_can_not_be_alone(self):

1425

options = self.parser.parse_args()

1426

options.all = True

1427

with self.assertParseError():

1428

self.check_option_syntax(options)

1429

1430

def test_all_is_ok_with_any_action(self):

1431

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

1432

options = self.parser.parse_args()

1433

setattr(options, action, value)

1434

options.all = True

1435

self.check_option_syntax(options)

1436

1437

def test_is_enabled_fails_without_client(self):

1438

options = self.parser.parse_args()

1439

options.is_enabled = True

1440

with self.assertParseError():

1441

self.check_option_syntax(options)

1442

1443

def test_is_enabled_works_with_one_client(self):

1444

options = self.parser.parse_args()

1445

options.is_enabled = True

1446

options.client = ["foo"]

1447

self.check_option_syntax(options)

1448

1449

def test_is_enabled_fails_with_two_clients(self):

1450

options = self.parser.parse_args()

1451

options.is_enabled = True

1452

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

1453

with self.assertParseError():

1454

self.check_option_syntax(options)

1455

1456

1457

1458

def should_only_run_tests():

1459

parser = argparse.ArgumentParser(add_help=False)

1460

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

1461

args, unknown_args = parser.parse_known_args()

1462

run_tests = args.check

1463

if run_tests:

1464

# Remove --check argument from sys.argv

1465

sys.argv[1:] = unknown_args

1466

return run_tests

1467

1468

# Add all tests from doctest strings

1469

def load_tests(loader, tests, none):

1470

import doctest

1471

tests.addTests(doctest.DocTestSuite())

1472

return tests

1473

1474

if __name__ == "__main__":

1475

if should_only_run_tests():

1476

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

1477

unittest.main()

1478

else:

1479

main()