/mandos/trunk : revision 1039

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

48

import dbus

49

50

# Show warnings by default

51

if not sys.warnoptions:

52

import warnings

53

warnings.simplefilter("default")

54

55

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

56

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

57

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

58

59

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

60

61

if sys.version_info.major == 2:

62

str = unicode

63

64

locale.setlocale(locale.LC_ALL, "")

65

66

domain = "se.recompile"

67

busname = domain + ".Mandos"

68

server_path = "/"

69

server_interface = domain + ".Mandos"

70

client_interface = domain + ".Mandos.Client"

71

version = "1.8.3"

72

73

74

try:

75

dbus.OBJECT_MANAGER_IFACE

76

except AttributeError:

77

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

78

79

80

def milliseconds_to_string(ms):

81

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

82

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

83

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

84

hours=td.seconds // 3600,

85

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

86

seconds=td.seconds % 60))

87

88

89

def rfc3339_duration_to_delta(duration):

90

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

91

92

>>> rfc3339_duration_to_delta("P7D")

93

datetime.timedelta(7)

94

>>> rfc3339_duration_to_delta("PT60S")

95

datetime.timedelta(0, 60)

96

>>> rfc3339_duration_to_delta("PT60M")

97

datetime.timedelta(0, 3600)

98

>>> rfc3339_duration_to_delta("P60M")

99

datetime.timedelta(1680)

100

>>> rfc3339_duration_to_delta("PT24H")

101

datetime.timedelta(1)

102

>>> rfc3339_duration_to_delta("P1W")

103

datetime.timedelta(7)

104

>>> rfc3339_duration_to_delta("PT5M30S")

105

datetime.timedelta(0, 330)

106

>>> rfc3339_duration_to_delta("P1DT3M20S")

107

datetime.timedelta(1, 200)

108

>>> # Can not be empty:

109

>>> rfc3339_duration_to_delta("")

110

Traceback (most recent call last):

111

...

112

ValueError: Invalid RFC 3339 duration: u''

113

>>> # Must start with "P":

114

>>> rfc3339_duration_to_delta("1D")

115

Traceback (most recent call last):

116

...

117

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

118

>>> # Must use correct order

119

>>> rfc3339_duration_to_delta("PT1S2M")

120

Traceback (most recent call last):

121

...

122

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

123

>>> # Time needs time marker

124

>>> rfc3339_duration_to_delta("P1H2S")

125

Traceback (most recent call last):

126

...

127

ValueError: Invalid RFC 3339 duration: u'P1H2S'

128

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

129

>>> rfc3339_duration_to_delta("P1D2W")

130

Traceback (most recent call last):

131

...

132

ValueError: Invalid RFC 3339 duration: u'P1D2W'

133

>>> rfc3339_duration_to_delta("P2W2H")

134

Traceback (most recent call last):

135

...

136

ValueError: Invalid RFC 3339 duration: u'P2W2H'

137

"""

138

139

# Parsing an RFC 3339 duration with regular expressions is not

140

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

141

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

142

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

143

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

144

# avoid excessive use of external libraries.

145

146

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

147

Token = collections.namedtuple("Token", (

148

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

149

# a "group" containing digits

150

"value", # datetime.timedelta or None

151

"followers")) # Tokens valid after this token

152

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

153

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

154

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

155

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

156

datetime.timedelta(seconds=1),

157

frozenset((token_end, )))

158

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

159

datetime.timedelta(minutes=1),

160

frozenset((token_second, token_end)))

161

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

162

datetime.timedelta(hours=1),

163

frozenset((token_minute, token_end)))

164

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

165

None,

166

frozenset((token_hour, token_minute,

167

token_second)))

168

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

169

datetime.timedelta(days=1),

170

frozenset((token_time, token_end)))

171

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

172

datetime.timedelta(weeks=4),

173

frozenset((token_day, token_end)))

174

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

175

datetime.timedelta(weeks=52),

176

frozenset((token_month, token_end)))

177

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

178

datetime.timedelta(weeks=1),

179

frozenset((token_end, )))

180

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

181

frozenset((token_year, token_month,

182

token_day, token_time,

183

token_week)))

184

# Define starting values:

185

# Value so far

186

value = datetime.timedelta()

187

found_token = None

188

# Following valid tokens

189

followers = frozenset((token_duration, ))

190

# String left to parse

191

s = duration

192

# Loop until end token is found

193

while found_token is not token_end:

194

# Search for any currently valid tokens

195

for token in followers:

196

match = token.regexp.match(s)

197

if match is not None:

198

# Token found

199

if token.value is not None:

200

# Value found, parse digits

201

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

202

# Add to value so far

203

value += factor * token.value

204

# Strip token from string

205

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

206

# Go to found token

207

found_token = token

208

# Set valid next tokens

209

followers = found_token.followers

210

break

211

else:

212

# No currently valid tokens were found

213

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

214

.format(duration))

215

# End token found

216

return value

217

218

219

def string_to_delta(interval):

220

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

221

222

try:

223

return rfc3339_duration_to_delta(interval)

224

except ValueError as e:

225

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

226

' '.join(e.args))

227

return parse_pre_1_6_1_interval(interval)

228

229

230

def parse_pre_1_6_1_interval(interval):

231

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

232

and return a datetime.timedelta

233

234

>>> parse_pre_1_6_1_interval('7d')

235

datetime.timedelta(7)

236

>>> parse_pre_1_6_1_interval('60s')

237

datetime.timedelta(0, 60)

238

>>> parse_pre_1_6_1_interval('60m')

239

datetime.timedelta(0, 3600)

240

>>> parse_pre_1_6_1_interval('24h')

241

datetime.timedelta(1)

242

>>> parse_pre_1_6_1_interval('1w')

243

datetime.timedelta(7)

244

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

245

datetime.timedelta(0, 330)

246

>>> parse_pre_1_6_1_interval('')

247

datetime.timedelta(0)

248

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

249

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

250

datetime.timedelta(2, 480, 18000)

251

252

"""

253

254

value = datetime.timedelta(0)

255

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

256

257

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

258

if suffix == "d":

259

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

260

elif suffix == "s":

261

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

262

elif suffix == "m":

263

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

264

elif suffix == "h":

265

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

266

elif suffix == "w":

267

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

268

elif suffix == "":

269

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

270

return value

271

272

273

## Classes for commands.

274

275

# Abstract classes first

276

class Command(object):

277

"""Abstract class for commands"""

278

def run(self, mandos, clients):

279

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

280

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

281

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

282

self.mandos = mandos

283

for client, properties in clients.items():

284

self.run_on_one_client(client, properties)

285

286

class PrintCmd(Command):

287

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

288

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

289

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

290

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

291

"ApprovalPending", "ApprovedByDefault",

292

"LastApprovalRequest", "ApprovalDelay",

293

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

294

"Expires", "LastCheckerStatus")

295

def run(self, mandos, clients):

296

print(self.output(clients))

297

298

class PropertyCmd(Command):

299

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

300

def run_on_one_client(self, client, properties):

301

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

302

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

303

dbus_interface=dbus.PROPERTIES_IFACE)

304

305

class ValueArgumentMixIn(object):

306

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

307

def __init__(self, value):

308

self.value_to_set = value

309

310

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

311

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

312

milliseconds."""

313

@property

314

def value_to_set(self):

315

return self._vts

316

@value_to_set.setter

317

def value_to_set(self, value):

318

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

319

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

320

321

# Actual (non-abstract) command classes

322

323

class PrintTableCmd(PrintCmd):

324

def __init__(self, verbose=False):

325

self.verbose = verbose

326

327

def output(self, clients):

328

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

329

keywords = default_keywords

330

if self.verbose:

331

keywords = self.all_keywords

332

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

333

334

class TableOfClients(object):

335

tableheaders = {

336

"Name": "Name",

337

"Enabled": "Enabled",

338

"Timeout": "Timeout",

339

"LastCheckedOK": "Last Successful Check",

340

"LastApprovalRequest": "Last Approval Request",

341

"Created": "Created",

342

"Interval": "Interval",

343

"Host": "Host",

344

"Fingerprint": "Fingerprint",

345

"KeyID": "Key ID",

346

"CheckerRunning": "Check Is Running",

347

"LastEnabled": "Last Enabled",

348

"ApprovalPending": "Approval Is Pending",

349

"ApprovedByDefault": "Approved By Default",

350

"ApprovalDelay": "Approval Delay",

351

"ApprovalDuration": "Approval Duration",

352

"Checker": "Checker",

353

"ExtendedTimeout": "Extended Timeout",

354

"Expires": "Expires",

355

"LastCheckerStatus": "Last Checker Status",

356

}

357

358

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

359

self.clients = clients

360

self.keywords = keywords

361

if tableheaders is not None:

362

self.tableheaders = tableheaders

363

364

def __str__(self):

365

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

366

367

if sys.version_info.major == 2:

368

__unicode__ = __str__

369

def __str__(self):

370

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

371

372

def rows(self):

373

format_string = self.row_formatting_string()

374

rows = [self.header_line(format_string)]

375

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

376

for client in self.clients)

377

return rows

378

379

def row_formatting_string(self):

380

"Format string used to format table rows"

381

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

382

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

383

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

384

for client in self.clients)),

385

key=key)

386

for key in self.keywords)

387

388

def string_from_client(self, client, key):

389

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

390

391

@staticmethod

392

def valuetostring(value, keyword):

393

if isinstance(value, dbus.Boolean):

394

return "Yes" if value else "No"

395

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

396

"ApprovalDuration", "ExtendedTimeout"):

397

return milliseconds_to_string(value)

398

return str(value)

399

400

def header_line(self, format_string):

401

return format_string.format(**self.tableheaders)

402

403

def client_line(self, client, format_string):

404

return format_string.format(

405

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

406

for key in self.keywords})

407

408

409

410

class DumpJSONCmd(PrintCmd):

411

def output(self, clients):

412

data = {client["Name"]:

413

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

414

for key in self.all_keywords}

415

for client in clients.values()}

416

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

417

@staticmethod

418

def dbus_boolean_to_bool(value):

419

if isinstance(value, dbus.Boolean):

420

value = bool(value)

421

return value

422

423

class IsEnabledCmd(Command):

424

def run_on_one_client(self, client, properties):

425

if self.is_enabled(client, properties):

426

sys.exit(0)

427

sys.exit(1)

428

def is_enabled(self, client, properties):

429

return bool(properties["Enabled"])

430

431

class RemoveCmd(Command):

432

def run_on_one_client(self, client, properties):

433

self.mandos.RemoveClient(client.__dbus_object_path__)

434

435

class ApproveCmd(Command):

436

def run_on_one_client(self, client, properties):

437

client.Approve(dbus.Boolean(True),

438

dbus_interface=client_interface)

439

440

class DenyCmd(Command):

441

def run_on_one_client(self, client, properties):

442

client.Approve(dbus.Boolean(False),

443

dbus_interface=client_interface)

444

445

class EnableCmd(PropertyCmd):

446

property = "Enabled"

447

value_to_set = dbus.Boolean(True)

448

449

class DisableCmd(PropertyCmd):

450

property = "Enabled"

451

value_to_set = dbus.Boolean(False)

452

453

class BumpTimeoutCmd(PropertyCmd):

454

property = "LastCheckedOK"

455

value_to_set = ""

456

457

class StartCheckerCmd(PropertyCmd):

458

property = "CheckerRunning"

459

value_to_set = dbus.Boolean(True)

460

461

class StopCheckerCmd(PropertyCmd):

462

property = "CheckerRunning"

463

value_to_set = dbus.Boolean(False)

464

465

class ApproveByDefaultCmd(PropertyCmd):

466

property = "ApprovedByDefault"

467

value_to_set = dbus.Boolean(True)

468

469

class DenyByDefaultCmd(PropertyCmd):

470

property = "ApprovedByDefault"

471

value_to_set = dbus.Boolean(False)

472

473

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

474

property = "Checker"

475

476

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

477

property = "Host"

478

479

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

480

@property

481

def value_to_set(self):

482

return self._vts

483

@value_to_set.setter

484

def value_to_set(self, value):

485

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

486

self._vts = value.read()

487

value.close()

488

property = "Secret"

489

490

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

491

property = "Timeout"

492

493

class SetExtendedTimeoutCmd(PropertyCmd,

494

MillisecondsValueArgumentMixIn):

495

property = "ExtendedTimeout"

496

497

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

498

property = "Interval"

499

500

class SetApprovalDelayCmd(PropertyCmd,

501

MillisecondsValueArgumentMixIn):

502

property = "ApprovalDelay"

503

504

class SetApprovalDurationCmd(PropertyCmd,

505

MillisecondsValueArgumentMixIn):

506

property = "ApprovalDuration"

507

508

def add_command_line_options(parser):

509

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

510

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

511

help="show version number and exit")

512

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

513

help="Select all clients")

514

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

515

help="Print all fields")

516

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

517

help="Dump client data in JSON format")

518

enable_disable = parser.add_mutually_exclusive_group()

519

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

520

help="Enable client")

521

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

522

action="store_true",

523

help="disable client")

524

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

525

help="Bump timeout for client")

526

start_stop_checker = parser.add_mutually_exclusive_group()

527

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

528

action="store_true",

529

help="Start checker for client")

530

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

531

action="store_true",

532

help="Stop checker for client")

533

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

534

help="Check if client is enabled")

535

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

536

help="Remove client")

537

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

538

help="Set checker command for client")

539

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

540

help="Set timeout for client")

541

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

542

help="Set extended timeout for client")

543

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

544

help="Set checker interval for client")

545

approve_deny_default = parser.add_mutually_exclusive_group()

546

approve_deny_default.add_argument(

547

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

548

default=None, dest="approved_by_default",

549

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

550

approve_deny_default.add_argument(

551

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

552

dest="approved_by_default",

553

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

554

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

555

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

556

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

557

help="Set duration of one client approval")

558

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

559

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

560

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

561

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

562

approve_deny = parser.add_mutually_exclusive_group()

563

approve_deny.add_argument(

564

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

565

help="Approve any current client request")

566

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

567

help="Deny any current client request")

568

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

569

help="Run self-test")

570

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

571

572

573

def commands_from_options(options):

574

575

commands = []

576

577

if options.dump_json:

578

commands.append(DumpJSONCmd())

579

580

if options.enable:

581

commands.append(EnableCmd())

582

583

if options.disable:

584

commands.append(DisableCmd())

585

586

if options.bump_timeout:

587

commands.append(BumpTimeoutCmd())

588

589

if options.start_checker:

590

commands.append(StartCheckerCmd())

591

592

if options.stop_checker:

593

commands.append(StopCheckerCmd())

594

595

if options.is_enabled:

596

commands.append(IsEnabledCmd())

597

598

if options.remove:

599

commands.append(RemoveCmd())

600

601

if options.checker is not None:

602

commands.append(SetCheckerCmd(options.checker))

603

604

if options.timeout is not None:

605

commands.append(SetTimeoutCmd(options.timeout))

606

607

if options.extended_timeout:

608

commands.append(

609

SetExtendedTimeoutCmd(options.extended_timeout))

610

611

if options.interval is not None:

612

commands.append(SetIntervalCmd(options.interval))

613

614

if options.approved_by_default is not None:

615

if options.approved_by_default:

616

commands.append(ApproveByDefaultCmd())

617

else:

618

commands.append(DenyByDefaultCmd())

619

620

if options.approval_delay is not None:

621

commands.append(SetApprovalDelayCmd(options.approval_delay))

622

623

if options.approval_duration is not None:

624

commands.append(

625

SetApprovalDurationCmd(options.approval_duration))

626

627

if options.host is not None:

628

commands.append(SetHostCmd(options.host))

629

630

if options.secret is not None:

631

commands.append(SetSecretCmd(options.secret))

632

633

if options.approve:

634

commands.append(ApproveCmd())

635

636

if options.deny:

637

commands.append(DenyCmd())

638

639

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

640

# optionally verbosely

641

if not commands:

642

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

643

644

return commands

645

646

647

def check_option_syntax(parser, options):

648

649

def has_actions(options):

650

return any((options.enable,

651

options.disable,

652

options.bump_timeout,

653

options.start_checker,

654

options.stop_checker,

655

options.is_enabled,

656

options.remove,

657

options.checker is not None,

658

options.timeout is not None,

659

options.extended_timeout is not None,

660

options.interval is not None,

661

options.approved_by_default is not None,

662

options.approval_delay is not None,

663

options.approval_duration is not None,

664

options.host is not None,

665

options.secret is not None,

666

options.approve,

667

options.deny))

668

669

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

670

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

671

if options.verbose and has_actions(options):

672

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

673

if options.dump_json and (options.verbose

674

or has_actions(options)):

675

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

676

if options.all and not has_actions(options):

677

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

678

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

679

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

680

681

682

def main():

683

parser = argparse.ArgumentParser()

684

685

add_command_line_options(parser)

686

687

options = parser.parse_args()

688

689

check_option_syntax(parser, options)

690

691

clientnames = options.client

692

693

try:

694

bus = dbus.SystemBus()

695

mandos_dbus_objc = bus.get_object(busname, server_path)

696

except dbus.exceptions.DBusException:

697

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

698

sys.exit(1)

699

700

mandos_serv = dbus.Interface(mandos_dbus_objc,

701

dbus_interface=server_interface)

702

mandos_serv_object_manager = dbus.Interface(

703

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

704

705

# Filter out log message from dbus module

706

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

707

class NullFilter(logging.Filter):

708

def filter(self, record):

709

return False

710

dbus_filter = NullFilter()

711

try:

712

dbus_logger.addFilter(dbus_filter)

713

mandos_clients = {path: ifs_and_props[client_interface]

714

for path, ifs_and_props in

715

mandos_serv_object_manager

716

.GetManagedObjects().items()

717

if client_interface in ifs_and_props}

718

except dbus.exceptions.DBusException as e:

719

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

720

"\n%s", e)

721

sys.exit(1)

722

finally:

723

# restore dbus logger

724

dbus_logger.removeFilter(dbus_filter)

725

726

# Compile dict of (clients: properties) to process

727

clients = {}

728

729

if not clientnames:

730

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

731

for path, properties in mandos_clients.items()}

732

else:

733

for name in clientnames:

734

for path, client in mandos_clients.items():

735

if client["Name"] == name:

736

client_objc = bus.get_object(busname, path)

737

clients[client_objc] = client

738

break

739

else:

740

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

741

sys.exit(1)

742

743

# Run all commands on clients

744

commands = commands_from_options(options)

745

for command in commands:

746

command.run(mandos_serv, clients)

747

748

749

class Test_milliseconds_to_string(unittest.TestCase):

750

def test_all(self):

751

self.assertEqual(milliseconds_to_string(93785000),

752

"1T02:03:05")

753

def test_no_days(self):

754

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

755

def test_all_zero(self):

756

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

757

def test_no_fractional_seconds(self):

758

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

759

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

760

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

761

762

class Test_string_to_delta(unittest.TestCase):

763

def test_handles_basic_rfc3339(self):

764

self.assertEqual(string_to_delta("PT0S"),

765

datetime.timedelta())

766

self.assertEqual(string_to_delta("P0D"),

767

datetime.timedelta())

768

self.assertEqual(string_to_delta("PT1S"),

769

datetime.timedelta(0, 1))

770

self.assertEqual(string_to_delta("PT2H"),

771

datetime.timedelta(0, 7200))

772

def test_falls_back_to_pre_1_6_1_with_warning(self):

773

# assertLogs only exists in Python 3.4

774

if hasattr(self, "assertLogs"):

775

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

776

value = string_to_delta("2h")

777

else:

778

class WarningFilter(logging.Filter):

779

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

780

def filter(self, record):

781

is_warning = record.levelno >= logging.WARNING

782

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

783

False)

784

return not is_warning

785

warning_filter = WarningFilter()

786

log.addFilter(warning_filter)

787

try:

788

value = string_to_delta("2h")

789

finally:

790

log.removeFilter(warning_filter)

791

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

792

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

793

794

795

class TestCmd(unittest.TestCase):

796

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

797

def setUp(self):

798

testcase = self

799

class MockClient(object):

800

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

801

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

802

self.attributes = attributes

803

self.attributes["Name"] = name

804

self.calls = []

805

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

806

testcase.assertEqual(interface, client_interface)

807

testcase.assertEqual(dbus_interface,

808

dbus.PROPERTIES_IFACE)

809

self.attributes[property] = value

810

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

811

testcase.assertEqual(interface, client_interface)

812

testcase.assertEqual(dbus_interface,

813

dbus.PROPERTIES_IFACE)

814

return self.attributes[property]

815

def Approve(self, approve, dbus_interface):

816

testcase.assertEqual(dbus_interface, client_interface)

817

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

818

dbus_interface)))

819

self.client = MockClient(

820

"foo",

821

KeyID=("92ed150794387c03ce684574b1139a65"

822

"94a34f895daaaf09fd8ea90a27cddb12"),

823

Secret=b"secret",

824

Host="foo.example.org",

825

Enabled=dbus.Boolean(True),

826

Timeout=300000,

827

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

828

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

829

Interval=120000,

830

Fingerprint=("778827225BA7DE539C5A"

831

"7CFA59CFF7CDBD9A5920"),

832

CheckerRunning=dbus.Boolean(False),

833

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

834

ApprovalPending=dbus.Boolean(False),

835

ApprovedByDefault=dbus.Boolean(True),

836

LastApprovalRequest="",

837

ApprovalDelay=0,

838

ApprovalDuration=1000,

839

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

840

ExtendedTimeout=900000,

841

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

842

LastCheckerStatus=0)

843

self.other_client = MockClient(

844

"barbar",

845

KeyID=("0558568eedd67d622f5c83b35a115f79"

846

"6ab612cff5ad227247e46c2b020f441c"),

847

Secret=b"secretbar",

848

Host="192.0.2.3",

849

Enabled=dbus.Boolean(True),

850

Timeout=300000,

851

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

852

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

853

Interval=120000,

854

Fingerprint=("3E393AEAEFB84C7E89E2"

855

"F547B3A107558FCA3A27"),

856

CheckerRunning=dbus.Boolean(True),

857

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

858

ApprovalPending=dbus.Boolean(False),

859

ApprovedByDefault=dbus.Boolean(False),

860

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

861

ApprovalDelay=30000,

862

ApprovalDuration=1000,

863

Checker=":",

864

ExtendedTimeout=900000,

865

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

866

LastCheckerStatus=-2)

867

self.clients = collections.OrderedDict(

868

[

869

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

870

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

871

])

872

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

873

874

class TestPrintTableCmd(TestCmd):

875

def test_normal(self):

876

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

877

expected_output = """

878

Name Enabled Timeout Last Successful Check

879

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

880

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

881

"""[1:-1]

882

self.assertEqual(output, expected_output)

883

def test_verbose(self):

884

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

885

expected_output = """

886

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

887

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

888

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

889

"""[1:-1]

890

self.assertEqual(output, expected_output)

891

def test_one_client(self):

892

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

893

expected_output = """

894

Name Enabled Timeout Last Successful Check

895

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

896

"""[1:-1]

897

self.assertEqual(output, expected_output)

898

899

class TestDumpJSONCmd(TestCmd):

900

def setUp(self):

901

self.expected_json = {

902

"foo": {

903

"Name": "foo",

904

"KeyID": ("92ed150794387c03ce684574b1139a65"

905

"94a34f895daaaf09fd8ea90a27cddb12"),

906

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

907

"Enabled": True,

908

"Timeout": 300000,

909

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

910

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

911

"Interval": 120000,

912

"Fingerprint": ("778827225BA7DE539C5A"

913

"7CFA59CFF7CDBD9A5920"),

914

"CheckerRunning": False,

915

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

916

"ApprovalPending": False,

917

"ApprovedByDefault": True,

918

"LastApprovalRequest": "",

919

"ApprovalDelay": 0,

920

"ApprovalDuration": 1000,

921

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

922

"ExtendedTimeout": 900000,

923

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

924

"LastCheckerStatus": 0,

925

},

926

"barbar": {

927

"Name": "barbar",

928

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

929

"6ab612cff5ad227247e46c2b020f441c"),

930

"Host": "192.0.2.3",

931

"Enabled": True,

932

"Timeout": 300000,

933

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

934

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

935

"Interval": 120000,

936

"Fingerprint": ("3E393AEAEFB84C7E89E2"

937

"F547B3A107558FCA3A27"),

938

"CheckerRunning": True,

939

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

940

"ApprovalPending": False,

941

"ApprovedByDefault": False,

942

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

943

"ApprovalDelay": 30000,

944

"ApprovalDuration": 1000,

945

"Checker": ":",

946

"ExtendedTimeout": 900000,

947

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

948

"LastCheckerStatus": -2,

949

},

950

}

951

return super(TestDumpJSONCmd, self).setUp()

952

def test_normal(self):

953

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

954

self.assertDictEqual(json_data, self.expected_json)

955

def test_one_client(self):

956

clients = self.one_client

957

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

958

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

959

self.assertDictEqual(json_data, expected_json)

960

961

class TestIsEnabledCmd(TestCmd):

962

def test_is_enabled(self):

963

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

964

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

965

def test_is_enabled_run_exits_successfully(self):

966

with self.assertRaises(SystemExit) as e:

967

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

968

if e.exception.code is not None:

969

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

970

else:

971

self.assertIsNone(e.exception.code)

972

def test_is_enabled_run_exits_with_failure(self):

973

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

974

with self.assertRaises(SystemExit) as e:

975

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

976

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

977

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

978

else:

979

self.assertIsNotNone(e.exception.code)

980

981

class TestRemoveCmd(TestCmd):

982

def test_remove(self):

983

class MockMandos(object):

984

def __init__(self):

985

self.calls = []

986

def RemoveClient(self, dbus_path):

987

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

988

mandos = MockMandos()

989

super(TestRemoveCmd, self).setUp()

990

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

991

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

992

for client in self.clients:

993

self.assertIn(("RemoveClient",

994

(client.__dbus_object_path__,)),

995

mandos.calls)

996

997

class TestApproveCmd(TestCmd):

998

def test_approve(self):

999

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

1000

for client in self.clients:

1001

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

1002

client.calls)

1003

1004

class TestDenyCmd(TestCmd):

1005

def test_deny(self):

1006

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

1007

for client in self.clients:

1008

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

1009

client.calls)

1010

1011

class TestEnableCmd(TestCmd):

1012

def test_enable(self):

1013

for client in self.clients:

1014

client.attributes["Enabled"] = False

1015

1016

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

1017

1018

for client in self.clients:

1019

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

1020

1021

class TestDisableCmd(TestCmd):

1022

def test_disable(self):

1023

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

1024

1025

for client in self.clients:

1026

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

1027

1028

class Unique(object):

1029

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

1030

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

1031

1032

class TestPropertyCmd(TestCmd):

1033

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

1034

def runTest(self):

1035

if not hasattr(self, "command"):

1036

return

1037

values_to_get = getattr(self, "values_to_get",

1038

self.values_to_set)

1039

for value_to_set, value_to_get in zip(self.values_to_set,

1040

values_to_get):

1041

for client in self.clients:

1042

old_value = client.attributes[self.property]

1043

self.assertNotIsInstance(old_value, Unique)

1044

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

1045

self.run_command(value_to_set, self.clients)

1046

for client in self.clients:

1047

value = client.attributes[self.property]

1048

self.assertNotIsInstance(value, Unique)

1049

self.assertEqual(value, value_to_get)

1050

def run_command(self, value, clients):

1051

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

1052

1053

class TestBumpTimeoutCmd(TestPropertyCmd):

1054

command = BumpTimeoutCmd

1055

property = "LastCheckedOK"

1056

values_to_set = [""]

1057

1058

class TestStartCheckerCmd(TestPropertyCmd):

1059

command = StartCheckerCmd

1060

property = "CheckerRunning"

1061

values_to_set = [dbus.Boolean(True)]

1062

1063

class TestStopCheckerCmd(TestPropertyCmd):

1064

command = StopCheckerCmd

1065

property = "CheckerRunning"

1066

values_to_set = [dbus.Boolean(False)]

1067

1068

class TestApproveByDefaultCmd(TestPropertyCmd):

1069

command = ApproveByDefaultCmd

1070

property = "ApprovedByDefault"

1071

values_to_set = [dbus.Boolean(True)]

1072

1073

class TestDenyByDefaultCmd(TestPropertyCmd):

1074

command = DenyByDefaultCmd

1075

property = "ApprovedByDefault"

1076

values_to_set = [dbus.Boolean(False)]

1077

1078

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1079

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

1080

ValueArgumentMixIn"""

1081

def runTest(self):

1082

if type(self) is TestValueArgumentPropertyCmd:

1083

return

1084

return super(TestValueArgumentPropertyCmd, self).runTest()

1085

def run_command(self, value, clients):

1086

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

1087

1088

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1089

command = SetCheckerCmd

1090

property = "Checker"

1091

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

1092

1093

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1094

command = SetHostCmd

1095

property = "Host"

1096

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

1097

1098

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1099

command = SetSecretCmd

1100

property = "Secret"

1101

values_to_set = [open("/dev/null", "rb"),

1102

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

1103

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

1104

1105

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1106

command = SetTimeoutCmd

1107

property = "Timeout"

1108

values_to_set = [datetime.timedelta(),

1109

datetime.timedelta(minutes=5),

1110

datetime.timedelta(seconds=1),

1111

datetime.timedelta(weeks=1),

1112

datetime.timedelta(weeks=52)]

1113

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

1114

1115

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1116

command = SetExtendedTimeoutCmd

1117

property = "ExtendedTimeout"

1118

values_to_set = [datetime.timedelta(),

1119

datetime.timedelta(minutes=5),

1120

datetime.timedelta(seconds=1),

1121

datetime.timedelta(weeks=1),

1122

datetime.timedelta(weeks=52)]

1123

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

1124

1125

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1126

command = SetIntervalCmd

1127

property = "Interval"

1128

values_to_set = [datetime.timedelta(),

1129

datetime.timedelta(minutes=5),

1130

datetime.timedelta(seconds=1),

1131

datetime.timedelta(weeks=1),

1132

datetime.timedelta(weeks=52)]

1133

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

1134

1135

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1136

command = SetApprovalDelayCmd

1137

property = "ApprovalDelay"

1138

values_to_set = [datetime.timedelta(),

1139

datetime.timedelta(minutes=5),

1140

datetime.timedelta(seconds=1),

1141

datetime.timedelta(weeks=1),

1142

datetime.timedelta(weeks=52)]

1143

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

1144

1145

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1146

command = SetApprovalDurationCmd

1147

property = "ApprovalDuration"

1148

values_to_set = [datetime.timedelta(),

1149

datetime.timedelta(minutes=5),

1150

datetime.timedelta(seconds=1),

1151

datetime.timedelta(weeks=1),

1152

datetime.timedelta(weeks=52)]

1153

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

1154

1155

class Test_command_from_options(unittest.TestCase):

1156

def setUp(self):

1157

self.parser = argparse.ArgumentParser()

1158

add_command_line_options(self.parser)

1159

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

1160

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

1161

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

1162

options = self.parser.parse_args(args)

1163

check_option_syntax(self.parser, options)

1164

commands = commands_from_options(options)

1165

self.assertEqual(len(commands), 1)

1166

command = commands[0]

1167

self.assertIsInstance(command, command_cls)

1168

for key, value in cmd_attrs.items():

1169

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

1170

def test_print_table(self):

1171

self.assert_command_from_args([], PrintTableCmd,

1172

verbose=False)

1173

1174

def test_print_table_verbose(self):

1175

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

1176

verbose=True)

1177

1178

def test_print_table_verbose_short(self):

1179

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

1180

verbose=True)

1181

1182

def test_enable(self):

1183

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

1184

1185

def test_enable_short(self):

1186

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

1187

1188

def test_disable(self):

1189

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

1190

DisableCmd)

1191

1192

def test_disable_short(self):

1193

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

1194

1195

def test_bump_timeout(self):

1196

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

1197

BumpTimeoutCmd)

1198

1199

def test_bump_timeout_short(self):

1200

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

1201

1202

def test_start_checker(self):

1203

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

1204

StartCheckerCmd)

1205

1206

def test_stop_checker(self):

1207

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

1208

StopCheckerCmd)

1209

1210

def test_remove(self):

1211

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

1212

RemoveCmd)

1213

1214

def test_remove_short(self):

1215

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

1216

1217

def test_checker(self):

1218

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

1219

SetCheckerCmd, value_to_set=":")

1220

1221

def test_checker_empty(self):

1222

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

1223

SetCheckerCmd, value_to_set="")

1224

1225

def test_checker_short(self):

1226

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

1227

SetCheckerCmd, value_to_set=":")

1228

1229

def test_timeout(self):

1230

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

1231

SetTimeoutCmd,

1232

value_to_set=300000)

1233

1234

def test_timeout_short(self):

1235

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

1236

SetTimeoutCmd,

1237

value_to_set=300000)

1238

1239

def test_extended_timeout(self):

1240

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

1241

"foo"],

1242

SetExtendedTimeoutCmd,

1243

value_to_set=900000)

1244

1245

def test_interval(self):

1246

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

1247

SetIntervalCmd,

1248

value_to_set=120000)

1249

1250

def test_interval_short(self):

1251

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

1252

SetIntervalCmd,

1253

value_to_set=120000)

1254

1255

def test_approve_by_default(self):

1256

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

1257

ApproveByDefaultCmd)

1258

1259

def test_deny_by_default(self):

1260

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

1261

DenyByDefaultCmd)

1262

1263

def test_approval_delay(self):

1264

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

1265

"foo"], SetApprovalDelayCmd,

1266

value_to_set=30000)

1267

1268

def test_approval_duration(self):

1269

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

1270

"foo"], SetApprovalDurationCmd,

1271

value_to_set=1000)

1272

1273

def test_host(self):

1274

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

1275

"foo"], SetHostCmd,

1276

value_to_set="foo.example.org")

1277

1278

def test_host_short(self):

1279

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

1280

"foo"], SetHostCmd,

1281

value_to_set="foo.example.org")

1282

1283

def test_secret_devnull(self):

1284

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

1285

"foo"], SetSecretCmd,

1286

value_to_set=b"")

1287

1288

def test_secret_tempfile(self):

1289

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

1290

value = b"secret\0xyzzy\nbar"

1291

f.write(value)

1292

f.seek(0)

1293

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

1294

"foo"], SetSecretCmd,

1295

value_to_set=value)

1296

1297

def test_secret_devnull_short(self):

1298

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

1299

SetSecretCmd, value_to_set=b"")

1300

1301

def test_secret_tempfile_short(self):

1302

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

1303

value = b"secret\0xyzzy\nbar"

1304

f.write(value)

1305

f.seek(0)

1306

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

1307

SetSecretCmd,

1308

value_to_set=value)

1309

1310

def test_approve(self):

1311

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

1312

ApproveCmd)

1313

1314

def test_approve_short(self):

1315

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

1316

1317

def test_deny(self):

1318

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

1319

1320

def test_deny_short(self):

1321

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

1322

1323

def test_dump_json(self):

1324

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

1325

1326

def test_is_enabled(self):

1327

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

1328

IsEnabledCmd)

1329

1330

def test_is_enabled_short(self):

1331

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

1332

1333

1334

1335

def should_only_run_tests():

1336

parser = argparse.ArgumentParser(add_help=False)

1337

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

1338

args, unknown_args = parser.parse_known_args()

1339

run_tests = args.check

1340

if run_tests:

1341

# Remove --check argument from sys.argv

1342

sys.argv[1:] = unknown_args

1343

return run_tests

1344

1345

# Add all tests from doctest strings

1346

def load_tests(loader, tests, none):

1347

import doctest

1348

tests.addTests(doctest.DocTestSuite())

1349

return tests

1350

1351

if __name__ == "__main__":

1352

if should_only_run_tests():

1353

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

1354

unittest.main()

1355

else:

1356

main()