/mandos/trunk : revision 1049

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.values()))

298

def output(self, clients):

299

raise NotImplementedError()

300

301

class PropertyCmd(Command):

302

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

303

def run_on_one_client(self, client, properties):

304

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

305

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

306

client.__dbus_object_path__,

307

dbus.PROPERTIES_IFACE, client_interface,

308

self.propname, self.value_to_set

309

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

310

else bool(self.value_to_set))

311

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

312

dbus_interface=dbus.PROPERTIES_IFACE)

313

@property

314

def propname(self):

315

raise NotImplementedError()

316

317

class ValueArgumentMixIn(object):

318

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

319

def __init__(self, value):

320

self.value_to_set = value

321

322

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

323

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

324

milliseconds."""

325

@property

326

def value_to_set(self):

327

return self._vts

328

@value_to_set.setter

329

def value_to_set(self, value):

330

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

331

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

332

333

# Actual (non-abstract) command classes

334

335

class PrintTableCmd(PrintCmd):

336

def __init__(self, verbose=False):

337

self.verbose = verbose

338

339

def output(self, clients):

340

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

341

keywords = default_keywords

342

if self.verbose:

343

keywords = self.all_keywords

344

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

345

346

class TableOfClients(object):

347

tableheaders = {

348

"Name": "Name",

349

"Enabled": "Enabled",

350

"Timeout": "Timeout",

351

"LastCheckedOK": "Last Successful Check",

352

"LastApprovalRequest": "Last Approval Request",

353

"Created": "Created",

354

"Interval": "Interval",

355

"Host": "Host",

356

"Fingerprint": "Fingerprint",

357

"KeyID": "Key ID",

358

"CheckerRunning": "Check Is Running",

359

"LastEnabled": "Last Enabled",

360

"ApprovalPending": "Approval Is Pending",

361

"ApprovedByDefault": "Approved By Default",

362

"ApprovalDelay": "Approval Delay",

363

"ApprovalDuration": "Approval Duration",

364

"Checker": "Checker",

365

"ExtendedTimeout": "Extended Timeout",

366

"Expires": "Expires",

367

"LastCheckerStatus": "Last Checker Status",

368

}

369

370

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

371

self.clients = clients

372

self.keywords = keywords

373

if tableheaders is not None:

374

self.tableheaders = tableheaders

375

376

def __str__(self):

377

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

378

379

if sys.version_info.major == 2:

380

__unicode__ = __str__

381

def __str__(self):

382

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

383

384

def rows(self):

385

format_string = self.row_formatting_string()

386

rows = [self.header_line(format_string)]

387

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

388

for client in self.clients)

389

return rows

390

391

def row_formatting_string(self):

392

"Format string used to format table rows"

393

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

394

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

395

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

396

for client in self.clients)),

397

key=key)

398

for key in self.keywords)

399

400

def string_from_client(self, client, key):

401

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

402

403

@staticmethod

404

def valuetostring(value, keyword):

405

if isinstance(value, dbus.Boolean):

406

return "Yes" if value else "No"

407

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

408

"ApprovalDuration", "ExtendedTimeout"):

409

return milliseconds_to_string(value)

410

return str(value)

411

412

def header_line(self, format_string):

413

return format_string.format(**self.tableheaders)

414

415

def client_line(self, client, format_string):

416

return format_string.format(

417

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

418

for key in self.keywords})

419

420

421

422

class DumpJSONCmd(PrintCmd):

423

def output(self, clients):

424

data = {client["Name"]:

425

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

426

for key in self.all_keywords}

427

for client in clients.values()}

428

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

429

@staticmethod

430

def dbus_boolean_to_bool(value):

431

if isinstance(value, dbus.Boolean):

432

value = bool(value)

433

return value

434

435

class IsEnabledCmd(Command):

436

def run_on_one_client(self, client, properties):

437

if self.is_enabled(client, properties):

438

sys.exit(0)

439

sys.exit(1)

440

def is_enabled(self, client, properties):

441

return properties["Enabled"]

442

443

class RemoveCmd(Command):

444

def run_on_one_client(self, client, properties):

445

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

446

server_path, server_interface,

447

str(client.__dbus_object_path__))

448

self.mandos.RemoveClient(client.__dbus_object_path__)

449

450

class ApproveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

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

453

client.__dbus_object_path__, client_interface)

454

client.Approve(dbus.Boolean(True),

455

dbus_interface=client_interface)

456

457

class DenyCmd(Command):

458

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_interface)

461

client.Approve(dbus.Boolean(False),

462

dbus_interface=client_interface)

463

464

class EnableCmd(PropertyCmd):

465

propname = "Enabled"

466

value_to_set = dbus.Boolean(True)

467

468

class DisableCmd(PropertyCmd):

469

propname = "Enabled"

470

value_to_set = dbus.Boolean(False)

471

472

class BumpTimeoutCmd(PropertyCmd):

473

propname = "LastCheckedOK"

474

value_to_set = ""

475

476

class StartCheckerCmd(PropertyCmd):

477

propname = "CheckerRunning"

478

value_to_set = dbus.Boolean(True)

479

480

class StopCheckerCmd(PropertyCmd):

481

propname = "CheckerRunning"

482

value_to_set = dbus.Boolean(False)

483

484

class ApproveByDefaultCmd(PropertyCmd):

485

propname = "ApprovedByDefault"

486

value_to_set = dbus.Boolean(True)

487

488

class DenyByDefaultCmd(PropertyCmd):

489

propname = "ApprovedByDefault"

490

value_to_set = dbus.Boolean(False)

491

492

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

493

propname = "Checker"

494

495

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

496

propname = "Host"

497

498

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

499

propname = "Secret"

500

@property

501

def value_to_set(self):

502

return self._vts

503

@value_to_set.setter

504

def value_to_set(self, value):

505

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

506

self._vts = value.read()

507

value.close()

508

509

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

510

propname = "Timeout"

511

512

class SetExtendedTimeoutCmd(PropertyCmd,

513

MillisecondsValueArgumentMixIn):

514

propname = "ExtendedTimeout"

515

516

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

517

propname = "Interval"

518

519

class SetApprovalDelayCmd(PropertyCmd,

520

MillisecondsValueArgumentMixIn):

521

propname = "ApprovalDelay"

522

523

class SetApprovalDurationCmd(PropertyCmd,

524

MillisecondsValueArgumentMixIn):

525

propname = "ApprovalDuration"

526

527

def add_command_line_options(parser):

528

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

529

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

530

help="show version number and exit")

531

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

532

help="Select all clients")

533

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

534

help="Print all fields")

535

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

536

help="Dump client data in JSON format")

537

enable_disable = parser.add_mutually_exclusive_group()

538

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

539

help="Enable client")

540

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

541

action="store_true",

542

help="disable client")

543

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

544

help="Bump timeout for client")

545

start_stop_checker = parser.add_mutually_exclusive_group()

546

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

547

action="store_true",

548

help="Start checker for client")

549

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

550

action="store_true",

551

help="Stop checker for client")

552

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

553

help="Check if client is enabled")

554

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

555

help="Remove client")

556

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

557

help="Set checker command for client")

558

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

559

help="Set timeout for client")

560

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

561

help="Set extended timeout for client")

562

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

563

help="Set checker interval for client")

564

approve_deny_default = parser.add_mutually_exclusive_group()

565

approve_deny_default.add_argument(

566

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

567

default=None, dest="approved_by_default",

568

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

569

approve_deny_default.add_argument(

570

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

571

dest="approved_by_default",

572

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

573

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

574

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

575

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

576

help="Set duration of one client approval")

577

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

578

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

579

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

580

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

581

approve_deny = parser.add_mutually_exclusive_group()

582

approve_deny.add_argument(

583

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

584

help="Approve any current client request")

585

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

586

help="Deny any current client request")

587

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

588

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

589

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

590

help="Run self-test")

591

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

592

593

594

def commands_from_options(options):

595

596

commands = []

597

598

if options.dump_json:

599

commands.append(DumpJSONCmd())

600

601

if options.enable:

602

commands.append(EnableCmd())

603

604

if options.disable:

605

commands.append(DisableCmd())

606

607

if options.bump_timeout:

608

commands.append(BumpTimeoutCmd())

609

610

if options.start_checker:

611

commands.append(StartCheckerCmd())

612

613

if options.stop_checker:

614

commands.append(StopCheckerCmd())

615

616

if options.is_enabled:

617

commands.append(IsEnabledCmd())

618

619

if options.checker is not None:

620

commands.append(SetCheckerCmd(options.checker))

621

622

if options.timeout is not None:

623

commands.append(SetTimeoutCmd(options.timeout))

624

625

if options.extended_timeout:

626

commands.append(

627

SetExtendedTimeoutCmd(options.extended_timeout))

628

629

if options.interval is not None:

630

commands.append(SetIntervalCmd(options.interval))

631

632

if options.approved_by_default is not None:

633

if options.approved_by_default:

634

commands.append(ApproveByDefaultCmd())

635

else:

636

commands.append(DenyByDefaultCmd())

637

638

if options.approval_delay is not None:

639

commands.append(SetApprovalDelayCmd(options.approval_delay))

640

641

if options.approval_duration is not None:

642

commands.append(

643

SetApprovalDurationCmd(options.approval_duration))

644

645

if options.host is not None:

646

commands.append(SetHostCmd(options.host))

647

648

if options.secret is not None:

649

commands.append(SetSecretCmd(options.secret))

650

651

if options.approve:

652

commands.append(ApproveCmd())

653

654

if options.deny:

655

commands.append(DenyCmd())

656

657

if options.remove:

658

commands.append(RemoveCmd())

659

660

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

661

# optionally verbosely

662

if not commands:

663

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

664

665

return commands

666

667

668

def check_option_syntax(parser, options):

669

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

670

argparse"""

671

672

def has_actions(options):

673

return any((options.enable,

674

options.disable,

675

options.bump_timeout,

676

options.start_checker,

677

options.stop_checker,

678

options.is_enabled,

679

options.remove,

680

options.checker is not None,

681

options.timeout is not None,

682

options.extended_timeout is not None,

683

options.interval is not None,

684

options.approved_by_default is not None,

685

options.approval_delay is not None,

686

options.approval_duration is not None,

687

options.host is not None,

688

options.secret is not None,

689

options.approve,

690

options.deny))

691

692

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

693

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

694

if options.verbose and has_actions(options):

695

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

696

if options.dump_json and (options.verbose

697

or has_actions(options)):

698

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

699

if options.all and not has_actions(options):

700

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

701

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

702

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

703

if options.remove:

704

options.remove = False

705

if has_actions(options) and not options.deny:

706

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

707

options.remove = True

708

709

710

def main():

711

parser = argparse.ArgumentParser()

712

713

add_command_line_options(parser)

714

715

options = parser.parse_args()

716

717

check_option_syntax(parser, options)

718

719

clientnames = options.client

720

721

if options.debug:

722

log.setLevel(logging.DEBUG)

723

724

try:

725

bus = dbus.SystemBus()

726

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

727

server_path)

728

mandos_dbus_objc = bus.get_object(busname, server_path)

729

except dbus.exceptions.DBusException:

730

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

731

sys.exit(1)

732

733

mandos_serv = dbus.Interface(mandos_dbus_objc,

734

dbus_interface=server_interface)

735

mandos_serv_object_manager = dbus.Interface(

736

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

737

738

# Filter out log message from dbus module

739

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

740

class NullFilter(logging.Filter):

741

def filter(self, record):

742

return False

743

dbus_filter = NullFilter()

744

try:

745

dbus_logger.addFilter(dbus_filter)

746

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

747

server_path, dbus.OBJECT_MANAGER_IFACE)

748

mandos_clients = {path: ifs_and_props[client_interface]

749

for path, ifs_and_props in

750

mandos_serv_object_manager

751

.GetManagedObjects().items()

752

if client_interface in ifs_and_props}

753

except dbus.exceptions.DBusException as e:

754

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

755

"\n%s", e)

756

sys.exit(1)

757

finally:

758

# restore dbus logger

759

dbus_logger.removeFilter(dbus_filter)

760

761

# Compile dict of (clients: properties) to process

762

clients = {}

763

764

if not clientnames:

765

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

766

busname, str(path)) and False) or

767

bus.get_object(busname, path): properties

768

for path, properties in mandos_clients.items()}

769

else:

770

for name in clientnames:

771

for path, client in mandos_clients.items():

772

if client["Name"] == name:

773

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

774

busname, str(path))

775

client_objc = bus.get_object(busname, path)

776

clients[client_objc] = client

777

break

778

else:

779

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

780

sys.exit(1)

781

782

# Run all commands on clients

783

commands = commands_from_options(options)

784

for command in commands:

785

command.run(mandos_serv, clients)

786

787

788

class Test_milliseconds_to_string(unittest.TestCase):

789

def test_all(self):

790

self.assertEqual(milliseconds_to_string(93785000),

791

"1T02:03:05")

792

def test_no_days(self):

793

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

794

def test_all_zero(self):

795

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

796

def test_no_fractional_seconds(self):

797

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

798

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

799

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

800

801

class Test_string_to_delta(unittest.TestCase):

802

def test_handles_basic_rfc3339(self):

803

self.assertEqual(string_to_delta("PT0S"),

804

datetime.timedelta())

805

self.assertEqual(string_to_delta("P0D"),

806

datetime.timedelta())

807

self.assertEqual(string_to_delta("PT1S"),

808

datetime.timedelta(0, 1))

809

self.assertEqual(string_to_delta("PT2H"),

810

datetime.timedelta(0, 7200))

811

def test_falls_back_to_pre_1_6_1_with_warning(self):

812

# assertLogs only exists in Python 3.4

813

if hasattr(self, "assertLogs"):

814

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

815

value = string_to_delta("2h")

816

else:

817

class WarningFilter(logging.Filter):

818

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

819

def filter(self, record):

820

is_warning = record.levelno >= logging.WARNING

821

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

822

False)

823

return not is_warning

824

warning_filter = WarningFilter()

825

log.addFilter(warning_filter)

826

try:

827

value = string_to_delta("2h")

828

finally:

829

log.removeFilter(warning_filter)

830

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

831

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

832

833

834

class TestCmd(unittest.TestCase):

835

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

836

def setUp(self):

837

testcase = self

838

class MockClient(object):

839

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

840

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

841

self.attributes = attributes

842

self.attributes["Name"] = name

843

self.calls = []

844

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

845

testcase.assertEqual(interface, client_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[propname] = value

849

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

850

testcase.assertEqual(interface, client_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[propname]

854

def Approve(self, approve, dbus_interface):

855

testcase.assertEqual(dbus_interface, client_interface)

856

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

857

dbus_interface)))

858

self.client = MockClient(

859

"foo",

860

KeyID=("92ed150794387c03ce684574b1139a65"

861

"94a34f895daaaf09fd8ea90a27cddb12"),

862

Secret=b"secret",

863

Host="foo.example.org",

864

Enabled=dbus.Boolean(True),

865

Timeout=300000,

866

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

867

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

868

Interval=120000,

869

Fingerprint=("778827225BA7DE539C5A"

870

"7CFA59CFF7CDBD9A5920"),

871

CheckerRunning=dbus.Boolean(False),

872

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

873

ApprovalPending=dbus.Boolean(False),

874

ApprovedByDefault=dbus.Boolean(True),

875

LastApprovalRequest="",

876

ApprovalDelay=0,

877

ApprovalDuration=1000,

878

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

879

ExtendedTimeout=900000,

880

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

881

LastCheckerStatus=0)

882

self.other_client = MockClient(

883

"barbar",

884

KeyID=("0558568eedd67d622f5c83b35a115f79"

885

"6ab612cff5ad227247e46c2b020f441c"),

886

Secret=b"secretbar",

887

Host="192.0.2.3",

888

Enabled=dbus.Boolean(True),

889

Timeout=300000,

890

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

891

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

892

Interval=120000,

893

Fingerprint=("3E393AEAEFB84C7E89E2"

894

"F547B3A107558FCA3A27"),

895

CheckerRunning=dbus.Boolean(True),

896

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

897

ApprovalPending=dbus.Boolean(False),

898

ApprovedByDefault=dbus.Boolean(False),

899

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

900

ApprovalDelay=30000,

901

ApprovalDuration=1000,

902

Checker=":",

903

ExtendedTimeout=900000,

904

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

905

LastCheckerStatus=-2)

906

self.clients = collections.OrderedDict(

907

[

908

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

909

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

910

])

911

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

912

913

class TestPrintTableCmd(TestCmd):

914

def test_normal(self):

915

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

916

expected_output = """

917

Name Enabled Timeout Last Successful Check

918

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

919

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

920

"""[1:-1]

921

self.assertEqual(output, expected_output)

922

def test_verbose(self):

923

output = PrintTableCmd(verbose=True).output(

924

self.clients.values())

925

expected_output = """

926

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

927

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

928

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

929

"""[1:-1]

930

self.assertEqual(output, expected_output)

931

def test_one_client(self):

932

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

933

expected_output = """

934

Name Enabled Timeout Last Successful Check

935

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

936

"""[1:-1]

937

self.assertEqual(output, expected_output)

938

939

class TestDumpJSONCmd(TestCmd):

940

def setUp(self):

941

self.expected_json = {

942

"foo": {

943

"Name": "foo",

944

"KeyID": ("92ed150794387c03ce684574b1139a65"

945

"94a34f895daaaf09fd8ea90a27cddb12"),

946

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

947

"Enabled": True,

948

"Timeout": 300000,

949

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

950

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

951

"Interval": 120000,

952

"Fingerprint": ("778827225BA7DE539C5A"

953

"7CFA59CFF7CDBD9A5920"),

954

"CheckerRunning": False,

955

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

956

"ApprovalPending": False,

957

"ApprovedByDefault": True,

958

"LastApprovalRequest": "",

959

"ApprovalDelay": 0,

960

"ApprovalDuration": 1000,

961

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

962

"ExtendedTimeout": 900000,

963

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

964

"LastCheckerStatus": 0,

965

},

966

"barbar": {

967

"Name": "barbar",

968

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

969

"6ab612cff5ad227247e46c2b020f441c"),

970

"Host": "192.0.2.3",

971

"Enabled": True,

972

"Timeout": 300000,

973

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

974

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

975

"Interval": 120000,

976

"Fingerprint": ("3E393AEAEFB84C7E89E2"

977

"F547B3A107558FCA3A27"),

978

"CheckerRunning": True,

979

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

980

"ApprovalPending": False,

981

"ApprovedByDefault": False,

982

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

983

"ApprovalDelay": 30000,

984

"ApprovalDuration": 1000,

985

"Checker": ":",

986

"ExtendedTimeout": 900000,

987

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

988

"LastCheckerStatus": -2,

989

},

990

}

991

return super(TestDumpJSONCmd, self).setUp()

992

def test_normal(self):

993

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

994

self.assertDictEqual(json_data, self.expected_json)

995

def test_one_client(self):

996

clients = self.one_client

997

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

998

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

999

self.assertDictEqual(json_data, expected_json)

1000

1001

class TestIsEnabledCmd(TestCmd):

1002

def test_is_enabled(self):

1003

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

1004

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

1005

def test_is_enabled_run_exits_successfully(self):

1006

with self.assertRaises(SystemExit) as e:

1007

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

1008

if e.exception.code is not None:

1009

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

1010

else:

1011

self.assertIsNone(e.exception.code)

1012

def test_is_enabled_run_exits_with_failure(self):

1013

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

1014

with self.assertRaises(SystemExit) as e:

1015

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

1016

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

1017

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

1018

else:

1019

self.assertIsNotNone(e.exception.code)

1020

1021

class TestRemoveCmd(TestCmd):

1022

def test_remove(self):

1023

class MockMandos(object):

1024

def __init__(self):

1025

self.calls = []

1026

def RemoveClient(self, dbus_path):

1027

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

1028

mandos = MockMandos()

1029

super(TestRemoveCmd, self).setUp()

1030

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

1031

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

1032

for client in self.clients:

1033

self.assertIn(("RemoveClient",

1034

(client.__dbus_object_path__,)),

1035

mandos.calls)

1036

1037

class TestApproveCmd(TestCmd):

1038

def test_approve(self):

1039

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

1040

for client in self.clients:

1041

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

1042

client.calls)

1043

1044

class TestDenyCmd(TestCmd):

1045

def test_deny(self):

1046

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

1047

for client in self.clients:

1048

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

1049

client.calls)

1050

1051

class TestEnableCmd(TestCmd):

1052

def test_enable(self):

1053

for client in self.clients:

1054

client.attributes["Enabled"] = False

1055

1056

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

1057

1058

for client in self.clients:

1059

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

1060

1061

class TestDisableCmd(TestCmd):

1062

def test_disable(self):

1063

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

1064

1065

for client in self.clients:

1066

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

1067

1068

class Unique(object):

1069

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

1070

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

1071

1072

class TestPropertyCmd(TestCmd):

1073

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

1074

def runTest(self):

1075

if not hasattr(self, "command"):

1076

return

1077

values_to_get = getattr(self, "values_to_get",

1078

self.values_to_set)

1079

for value_to_set, value_to_get in zip(self.values_to_set,

1080

values_to_get):

1081

for client in self.clients:

1082

old_value = client.attributes[self.propname]

1083

self.assertNotIsInstance(old_value, Unique)

1084

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

1085

self.run_command(value_to_set, self.clients)

1086

for client in self.clients:

1087

value = client.attributes[self.propname]

1088

self.assertNotIsInstance(value, Unique)

1089

self.assertEqual(value, value_to_get)

1090

def run_command(self, value, clients):

1091

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

1092

1093

class TestBumpTimeoutCmd(TestPropertyCmd):

1094

command = BumpTimeoutCmd

1095

propname = "LastCheckedOK"

1096

values_to_set = [""]

1097

1098

class TestStartCheckerCmd(TestPropertyCmd):

1099

command = StartCheckerCmd

1100

propname = "CheckerRunning"

1101

values_to_set = [dbus.Boolean(True)]

1102

1103

class TestStopCheckerCmd(TestPropertyCmd):

1104

command = StopCheckerCmd

1105

propname = "CheckerRunning"

1106

values_to_set = [dbus.Boolean(False)]

1107

1108

class TestApproveByDefaultCmd(TestPropertyCmd):

1109

command = ApproveByDefaultCmd

1110

propname = "ApprovedByDefault"

1111

values_to_set = [dbus.Boolean(True)]

1112

1113

class TestDenyByDefaultCmd(TestPropertyCmd):

1114

command = DenyByDefaultCmd

1115

propname = "ApprovedByDefault"

1116

values_to_set = [dbus.Boolean(False)]

1117

1118

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1119

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

1120

ValueArgumentMixIn"""

1121

def runTest(self):

1122

if type(self) is TestValueArgumentPropertyCmd:

1123

return

1124

return super(TestValueArgumentPropertyCmd, self).runTest()

1125

def run_command(self, value, clients):

1126

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

1127

1128

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1129

command = SetCheckerCmd

1130

propname = "Checker"

1131

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

1132

1133

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1134

command = SetHostCmd

1135

propname = "Host"

1136

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

1137

1138

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1139

command = SetSecretCmd

1140

propname = "Secret"

1141

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

1142

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

1143

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

1144

1145

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1146

command = SetTimeoutCmd

1147

propname = "Timeout"

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 TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1156

command = SetExtendedTimeoutCmd

1157

propname = "ExtendedTimeout"

1158

values_to_set = [datetime.timedelta(),

1159

datetime.timedelta(minutes=5),

1160

datetime.timedelta(seconds=1),

1161

datetime.timedelta(weeks=1),

1162

datetime.timedelta(weeks=52)]

1163

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

1164

1165

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1166

command = SetIntervalCmd

1167

propname = "Interval"

1168

values_to_set = [datetime.timedelta(),

1169

datetime.timedelta(minutes=5),

1170

datetime.timedelta(seconds=1),

1171

datetime.timedelta(weeks=1),

1172

datetime.timedelta(weeks=52)]

1173

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

1174

1175

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1176

command = SetApprovalDelayCmd

1177

propname = "ApprovalDelay"

1178

values_to_set = [datetime.timedelta(),

1179

datetime.timedelta(minutes=5),

1180

datetime.timedelta(seconds=1),

1181

datetime.timedelta(weeks=1),

1182

datetime.timedelta(weeks=52)]

1183

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

1184

1185

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1186

command = SetApprovalDurationCmd

1187

propname = "ApprovalDuration"

1188

values_to_set = [datetime.timedelta(),

1189

datetime.timedelta(minutes=5),

1190

datetime.timedelta(seconds=1),

1191

datetime.timedelta(weeks=1),

1192

datetime.timedelta(weeks=52)]

1193

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

1194

1195

class Test_command_from_options(unittest.TestCase):

1196

def setUp(self):

1197

self.parser = argparse.ArgumentParser()

1198

add_command_line_options(self.parser)

1199

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

1200

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

1201

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

1202

options = self.parser.parse_args(args)

1203

check_option_syntax(self.parser, options)

1204

commands = commands_from_options(options)

1205

self.assertEqual(len(commands), 1)

1206

command = commands[0]

1207

self.assertIsInstance(command, command_cls)

1208

for key, value in cmd_attrs.items():

1209

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

1210

def test_print_table(self):

1211

self.assert_command_from_args([], PrintTableCmd,

1212

verbose=False)

1213

1214

def test_print_table_verbose(self):

1215

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

1216

verbose=True)

1217

1218

def test_print_table_verbose_short(self):

1219

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

1220

verbose=True)

1221

1222

def test_enable(self):

1223

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

1224

1225

def test_enable_short(self):

1226

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

1227

1228

def test_disable(self):

1229

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

1230

DisableCmd)

1231

1232

def test_disable_short(self):

1233

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

1234

1235

def test_bump_timeout(self):

1236

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

1237

BumpTimeoutCmd)

1238

1239

def test_bump_timeout_short(self):

1240

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

1241

1242

def test_start_checker(self):

1243

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

1244

StartCheckerCmd)

1245

1246

def test_stop_checker(self):

1247

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

1248

StopCheckerCmd)

1249

1250

def test_remove(self):

1251

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

1252

RemoveCmd)

1253

1254

def test_remove_short(self):

1255

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

1256

1257

def test_checker(self):

1258

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

1259

SetCheckerCmd, value_to_set=":")

1260

1261

def test_checker_empty(self):

1262

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

1263

SetCheckerCmd, value_to_set="")

1264

1265

def test_checker_short(self):

1266

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

1267

SetCheckerCmd, value_to_set=":")

1268

1269

def test_timeout(self):

1270

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

1271

SetTimeoutCmd,

1272

value_to_set=300000)

1273

1274

def test_timeout_short(self):

1275

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

1276

SetTimeoutCmd,

1277

value_to_set=300000)

1278

1279

def test_extended_timeout(self):

1280

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

1281

"foo"],

1282

SetExtendedTimeoutCmd,

1283

value_to_set=900000)

1284

1285

def test_interval(self):

1286

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

1287

SetIntervalCmd,

1288

value_to_set=120000)

1289

1290

def test_interval_short(self):

1291

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

1292

SetIntervalCmd,

1293

value_to_set=120000)

1294

1295

def test_approve_by_default(self):

1296

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

1297

ApproveByDefaultCmd)

1298

1299

def test_deny_by_default(self):

1300

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

1301

DenyByDefaultCmd)

1302

1303

def test_approval_delay(self):

1304

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

1305

"foo"], SetApprovalDelayCmd,

1306

value_to_set=30000)

1307

1308

def test_approval_duration(self):

1309

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

1310

"foo"], SetApprovalDurationCmd,

1311

value_to_set=1000)

1312

1313

def test_host(self):

1314

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

1315

"foo"], SetHostCmd,

1316

value_to_set="foo.example.org")

1317

1318

def test_host_short(self):

1319

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

1320

"foo"], SetHostCmd,

1321

value_to_set="foo.example.org")

1322

1323

def test_secret_devnull(self):

1324

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

1325

"foo"], SetSecretCmd,

1326

value_to_set=b"")

1327

1328

def test_secret_tempfile(self):

1329

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

1330

value = b"secret\0xyzzy\nbar"

1331

f.write(value)

1332

f.seek(0)

1333

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

1334

"foo"], SetSecretCmd,

1335

value_to_set=value)

1336

1337

def test_secret_devnull_short(self):

1338

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

1339

SetSecretCmd, value_to_set=b"")

1340

1341

def test_secret_tempfile_short(self):

1342

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

1343

value = b"secret\0xyzzy\nbar"

1344

f.write(value)

1345

f.seek(0)

1346

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

1347

SetSecretCmd,

1348

value_to_set=value)

1349

1350

def test_approve(self):

1351

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

1352

ApproveCmd)

1353

1354

def test_approve_short(self):

1355

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

1356

1357

def test_deny(self):

1358

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

1359

1360

def test_deny_short(self):

1361

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

1362

1363

def test_dump_json(self):

1364

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

1365

1366

def test_is_enabled(self):

1367

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

1368

IsEnabledCmd)

1369

1370

def test_is_enabled_short(self):

1371

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

1372

1373

def test_deny_before_remove(self):

1374

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

1375

check_option_syntax(self.parser, options)

1376

commands = commands_from_options(options)

1377

self.assertEqual(len(commands), 2)

1378

self.assertIsInstance(commands[0], DenyCmd)

1379

self.assertIsInstance(commands[1], RemoveCmd)

1380

1381

def test_deny_before_remove_reversed(self):

1382

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

1383

check_option_syntax(self.parser, options)

1384

commands = commands_from_options(options)

1385

self.assertEqual(len(commands), 2)

1386

self.assertIsInstance(commands[0], DenyCmd)

1387

self.assertIsInstance(commands[1], RemoveCmd)

1388

1389

1390

class Test_check_option_syntax(unittest.TestCase):

1391

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

1392

# check_option_syntax()

1393

actions = {

1394

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

1395

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

1396

# never used

1397

"enable": True,

1398

"disable": True,

1399

"bump_timeout": True,

1400

"start_checker": True,

1401

"stop_checker": True,

1402

"is_enabled": True,

1403

"remove": True,

1404

"checker": "x",

1405

"timeout": datetime.timedelta(),

1406

"extended_timeout": datetime.timedelta(),

1407

"interval": datetime.timedelta(),

1408

"approved_by_default": True,

1409

"approval_delay": datetime.timedelta(),

1410

"approval_duration": datetime.timedelta(),

1411

"host": "x",

1412

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

1413

"approve": True,

1414

"deny": True,

1415

}

1416

1417

def setUp(self):

1418

self.parser = argparse.ArgumentParser()

1419

add_command_line_options(self.parser)

1420

1421

@contextlib.contextmanager

1422

def assertParseError(self):

1423

with self.assertRaises(SystemExit) as e:

1424

with self.temporarily_suppress_stderr():

1425

yield

1426

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

1427

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

1428

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

1429

1430

@staticmethod

1431

@contextlib.contextmanager

1432

def temporarily_suppress_stderr():

1433

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

1434

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

1435

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

1436

os.close(null)

1437

try:

1438

yield

1439

finally:

1440

# restore stderr

1441

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

1442

os.close(stderrcopy)

1443

1444

def check_option_syntax(self, options):

1445

check_option_syntax(self.parser, options)

1446

1447

def test_actions_requires_client_or_all(self):

1448

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

1449

options = self.parser.parse_args()

1450

setattr(options, action, value)

1451

with self.assertParseError():

1452

self.check_option_syntax(options)

1453

1454

def test_actions_conflicts_with_verbose(self):

1455

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

1456

options = self.parser.parse_args()

1457

setattr(options, action, value)

1458

options.verbose = True

1459

with self.assertParseError():

1460

self.check_option_syntax(options)

1461

1462

def test_dump_json_conflicts_with_verbose(self):

1463

options = self.parser.parse_args()

1464

options.dump_json = True

1465

options.verbose = True

1466

with self.assertParseError():

1467

self.check_option_syntax(options)

1468

1469

def test_dump_json_conflicts_with_action(self):

1470

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

1471

options = self.parser.parse_args()

1472

setattr(options, action, value)

1473

options.dump_json = True

1474

with self.assertParseError():

1475

self.check_option_syntax(options)

1476

1477

def test_all_can_not_be_alone(self):

1478

options = self.parser.parse_args()

1479

options.all = True

1480

with self.assertParseError():

1481

self.check_option_syntax(options)

1482

1483

def test_all_is_ok_with_any_action(self):

1484

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

1485

options = self.parser.parse_args()

1486

setattr(options, action, value)

1487

options.all = True

1488

self.check_option_syntax(options)

1489

1490

def test_is_enabled_fails_without_client(self):

1491

options = self.parser.parse_args()

1492

options.is_enabled = True

1493

with self.assertParseError():

1494

self.check_option_syntax(options)

1495

1496

def test_is_enabled_works_with_one_client(self):

1497

options = self.parser.parse_args()

1498

options.is_enabled = True

1499

options.client = ["foo"]

1500

self.check_option_syntax(options)

1501

1502

def test_is_enabled_fails_with_two_clients(self):

1503

options = self.parser.parse_args()

1504

options.is_enabled = True

1505

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

1506

with self.assertParseError():

1507

self.check_option_syntax(options)

1508

1509

def test_remove_can_only_be_combined_with_action_deny(self):

1510

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

1511

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

1512

continue

1513

options = self.parser.parse_args()

1514

setattr(options, action, value)

1515

options.all = True

1516

options.remove = True

1517

with self.assertParseError():

1518

self.check_option_syntax(options)

1519

1520

1521

1522

def should_only_run_tests():

1523

parser = argparse.ArgumentParser(add_help=False)

1524

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

1525

args, unknown_args = parser.parse_known_args()

1526

run_tests = args.check

1527

if run_tests:

1528

# Remove --check argument from sys.argv

1529

sys.argv[1:] = unknown_args

1530

return run_tests

1531

1532

# Add all tests from doctest strings

1533

def load_tests(loader, tests, none):

1534

import doctest

1535

tests.addTests(doctest.DocTestSuite())

1536

return tests

1537

1538

if __name__ == "__main__":

1539

if should_only_run_tests():

1540

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

1541

unittest.main()

1542

else:

1543

main()