/mandos/release : revision 103

1

#!/usr/bin/python

2

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

# Mandos is free software: you can redistribute it and/or modify it

12

# under the terms of the GNU General Public License as published by

13

# the Free Software Foundation, either version 3 of the License, or

14

# (at your option) any later version.

15

#

16

# Mandos is distributed in the hope that it will be useful, but

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19

# GNU General Public License for more details.

20

#

21

# You should have received a copy of the GNU General Public License

22

# along with Mandos. If not, see <http://www.gnu.org/licenses/>.

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

35

import sys

36

import argparse

37

import locale

38

import datetime

39

import re

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

64

65

locale.setlocale(locale.LC_ALL, "")

66

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

81

82

def milliseconds_to_string(ms):

83

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

84

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

85

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

86

hours=td.seconds // 3600,

87

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

88

seconds=td.seconds % 60))

89

90

91

def rfc3339_duration_to_delta(duration):

92

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

93

94

>>> rfc3339_duration_to_delta("P7D")

95

datetime.timedelta(7)

96

>>> rfc3339_duration_to_delta("PT60S")

97

datetime.timedelta(0, 60)

98

>>> rfc3339_duration_to_delta("PT60M")

99

datetime.timedelta(0, 3600)

100

>>> rfc3339_duration_to_delta("P60M")

101

datetime.timedelta(1680)

102

>>> rfc3339_duration_to_delta("PT24H")

103

datetime.timedelta(1)

104

>>> rfc3339_duration_to_delta("P1W")

105

datetime.timedelta(7)

106

>>> rfc3339_duration_to_delta("PT5M30S")

107

datetime.timedelta(0, 330)

108

>>> rfc3339_duration_to_delta("P1DT3M20S")

109

datetime.timedelta(1, 200)

110

>>> # Can not be empty:

111

>>> rfc3339_duration_to_delta("")

112

Traceback (most recent call last):

113

...

114

ValueError: Invalid RFC 3339 duration: u''

115

>>> # Must start with "P":

116

>>> rfc3339_duration_to_delta("1D")

117

Traceback (most recent call last):

118

...

119

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

120

>>> # Must use correct order

121

>>> rfc3339_duration_to_delta("PT1S2M")

122

Traceback (most recent call last):

123

...

124

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

125

>>> # Time needs time marker

126

>>> rfc3339_duration_to_delta("P1H2S")

127

Traceback (most recent call last):

128

...

129

ValueError: Invalid RFC 3339 duration: u'P1H2S'

130

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

131

>>> rfc3339_duration_to_delta("P1D2W")

132

Traceback (most recent call last):

133

...

134

ValueError: Invalid RFC 3339 duration: u'P1D2W'

135

>>> rfc3339_duration_to_delta("P2W2H")

136

Traceback (most recent call last):

137

...

138

ValueError: Invalid RFC 3339 duration: u'P2W2H'

139

"""

140

141

# Parsing an RFC 3339 duration with regular expressions is not

142

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

143

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

144

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

145

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

146

# avoid excessive use of external libraries.

147

148

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

149

Token = collections.namedtuple("Token", (

150

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

151

# a "group" containing digits

152

"value", # datetime.timedelta or None

153

"followers")) # Tokens valid after this token

154

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

155

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

156

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

157

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

158

datetime.timedelta(seconds=1),

159

frozenset((token_end, )))

160

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

161

datetime.timedelta(minutes=1),

162

frozenset((token_second, token_end)))

163

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

164

datetime.timedelta(hours=1),

165

frozenset((token_minute, token_end)))

166

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

167

None,

168

frozenset((token_hour, token_minute,

169

token_second)))

170

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

171

datetime.timedelta(days=1),

172

frozenset((token_time, token_end)))

173

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

174

datetime.timedelta(weeks=4),

175

frozenset((token_day, token_end)))

176

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

177

datetime.timedelta(weeks=52),

178

frozenset((token_month, token_end)))

179

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

180

datetime.timedelta(weeks=1),

181

frozenset((token_end, )))

182

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

183

frozenset((token_year, token_month,

184

token_day, token_time,

185

token_week)))

186

# Define starting values:

187

# Value so far

188

value = datetime.timedelta()

189

found_token = None

190

# Following valid tokens

191

followers = frozenset((token_duration, ))

192

# String left to parse

193

s = duration

194

# Loop until end token is found

195

while found_token is not token_end:

196

# Search for any currently valid tokens

197

for token in followers:

198

match = token.regexp.match(s)

199

if match is not None:

200

# Token found

201

if token.value is not None:

202

# Value found, parse digits

203

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

204

# Add to value so far

205

value += factor * token.value

206

# Strip token from string

207

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

208

# Go to found token

209

found_token = token

210

# Set valid next tokens

211

followers = found_token.followers

212

break

213

else:

214

# No currently valid tokens were found

215

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

216

.format(duration))

217

# End token found

218

return value

219

220

221

def string_to_delta(interval):

222

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

223

224

try:

225

return rfc3339_duration_to_delta(interval)

226

except ValueError as e:

227

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

228

' '.join(e.args))

229

return parse_pre_1_6_1_interval(interval)

230

231

232

def parse_pre_1_6_1_interval(interval):

233

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

234

and return a datetime.timedelta

235

236

>>> parse_pre_1_6_1_interval('7d')

237

datetime.timedelta(7)

238

>>> parse_pre_1_6_1_interval('60s')

239

datetime.timedelta(0, 60)

240

>>> parse_pre_1_6_1_interval('60m')

241

datetime.timedelta(0, 3600)

242

>>> parse_pre_1_6_1_interval('24h')

243

datetime.timedelta(1)

244

>>> parse_pre_1_6_1_interval('1w')

245

datetime.timedelta(7)

246

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

247

datetime.timedelta(0, 330)

248

>>> parse_pre_1_6_1_interval('')

249

datetime.timedelta(0)

250

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

251

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

252

datetime.timedelta(2, 480, 18000)

253

254

"""

255

256

value = datetime.timedelta(0)

257

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

258

259

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

260

if suffix == "d":

261

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

262

elif suffix == "s":

263

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

264

elif suffix == "m":

265

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

266

elif suffix == "h":

267

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

268

elif suffix == "w":

269

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

270

elif suffix == "":

271

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

272

return value

273

274

275

## Classes for commands.

276

277

# Abstract classes first

278

class Command(object):

279

"""Abstract class for commands"""

280

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

281

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

282

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

283

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

284

self.mandos = mandos

285

for clientpath, properties in clients.items():

286

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

287

dbus_busname, str(clientpath))

288

client = bus.get_object(dbus_busname, clientpath)

289

self.run_on_one_client(client, properties)

290

291

class PrintCmd(Command):

292

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

293

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

294

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

295

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

296

"ApprovalPending", "ApprovedByDefault",

297

"LastApprovalRequest", "ApprovalDelay",

298

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

299

"Expires", "LastCheckerStatus")

300

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

301

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

302

def output(self, clients):

303

raise NotImplementedError()

304

305

class PropertyCmd(Command):

306

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

307

def run_on_one_client(self, client, properties):

308

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

309

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

310

client.__dbus_object_path__,

311

dbus.PROPERTIES_IFACE, client_dbus_interface,

312

self.propname, self.value_to_set

313

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

314

else bool(self.value_to_set))

315

client.Set(client_dbus_interface, self.propname,

316

self.value_to_set,

317

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

321

322

class PropertyValueCmd(PropertyCmd):

323

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

324

def __init__(self, value):

325

self.value_to_set = value

326

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

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

329

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

330

@property

331

def value_to_set(self):

332

return self._vts

333

@value_to_set.setter

334

def value_to_set(self, value):

335

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

336

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

337

338

# Actual (non-abstract) command classes

339

340

class PrintTableCmd(PrintCmd):

341

def __init__(self, verbose=False):

342

self.verbose = verbose

343

344

def output(self, clients):

345

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

346

"LastCheckedOK")

347

keywords = default_keywords

348

if self.verbose:

349

keywords = self.all_keywords

350

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

351

352

class TableOfClients(object):

353

tableheaders = {

354

"Name": "Name",

355

"Enabled": "Enabled",

356

"Timeout": "Timeout",

357

"LastCheckedOK": "Last Successful Check",

358

"LastApprovalRequest": "Last Approval Request",

359

"Created": "Created",

360

"Interval": "Interval",

361

"Host": "Host",

362

"Fingerprint": "Fingerprint",

363

"KeyID": "Key ID",

364

"CheckerRunning": "Check Is Running",

365

"LastEnabled": "Last Enabled",

366

"ApprovalPending": "Approval Is Pending",

367

"ApprovedByDefault": "Approved By Default",

368

"ApprovalDelay": "Approval Delay",

369

"ApprovalDuration": "Approval Duration",

370

"Checker": "Checker",

371

"ExtendedTimeout": "Extended Timeout",

372

"Expires": "Expires",

373

"LastCheckerStatus": "Last Checker Status",

374

}

375

376

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

377

self.clients = clients

378

self.keywords = keywords

379

if tableheaders is not None:

380

self.tableheaders = tableheaders

381

382

def __str__(self):

383

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

384

385

if sys.version_info.major == 2:

386

__unicode__ = __str__

387

def __str__(self):

388

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

389

390

def rows(self):

391

format_string = self.row_formatting_string()

392

rows = [self.header_line(format_string)]

393

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

394

for client in self.clients)

395

return rows

396

397

def row_formatting_string(self):

398

"Format string used to format table rows"

399

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

400

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

401

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

402

for client in self.clients)),

403

key=key)

404

for key in self.keywords)

405

406

def string_from_client(self, client, key):

407

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

408

409

@staticmethod

410

def valuetostring(value, keyword):

411

if isinstance(value, dbus.Boolean):

412

return "Yes" if value else "No"

413

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

414

"ApprovalDuration", "ExtendedTimeout"):

415

return milliseconds_to_string(value)

416

return str(value)

417

418

def header_line(self, format_string):

419

return format_string.format(**self.tableheaders)

420

421

def client_line(self, client, format_string):

422

return format_string.format(

423

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

424

for key in self.keywords})

425

426

427

428

class DumpJSONCmd(PrintCmd):

429

def output(self, clients):

430

data = {client["Name"]:

431

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

432

for key in self.all_keywords}

433

for client in clients.values()}

434

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

435

@staticmethod

436

def dbus_boolean_to_bool(value):

437

if isinstance(value, dbus.Boolean):

438

value = bool(value)

439

return value

440

441

class IsEnabledCmd(Command):

442

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

443

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

444

if self.is_enabled(client, properties):

445

sys.exit(0)

446

sys.exit(1)

447

def is_enabled(self, client, properties):

448

return properties["Enabled"]

449

450

class RemoveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

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

453

server_dbus_path, server_dbus_interface,

454

str(client.__dbus_object_path__))

455

self.mandos.RemoveClient(client.__dbus_object_path__)

456

457

class ApproveCmd(Command):

458

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_dbus_interface)

461

client.Approve(dbus.Boolean(True),

462

dbus_interface=client_dbus_interface)

463

464

class DenyCmd(Command):

465

def run_on_one_client(self, client, properties):

466

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

467

client.__dbus_object_path__, client_dbus_interface)

468

client.Approve(dbus.Boolean(False),

469

dbus_interface=client_dbus_interface)

470

471

class EnableCmd(PropertyCmd):

472

propname = "Enabled"

473

value_to_set = dbus.Boolean(True)

474

475

class DisableCmd(PropertyCmd):

476

propname = "Enabled"

477

value_to_set = dbus.Boolean(False)

478

479

class BumpTimeoutCmd(PropertyCmd):

480

propname = "LastCheckedOK"

481

value_to_set = ""

482

483

class StartCheckerCmd(PropertyCmd):

484

propname = "CheckerRunning"

485

value_to_set = dbus.Boolean(True)

486

487

class StopCheckerCmd(PropertyCmd):

488

propname = "CheckerRunning"

489

value_to_set = dbus.Boolean(False)

490

491

class ApproveByDefaultCmd(PropertyCmd):

492

propname = "ApprovedByDefault"

493

value_to_set = dbus.Boolean(True)

494

495

class DenyByDefaultCmd(PropertyCmd):

496

propname = "ApprovedByDefault"

497

value_to_set = dbus.Boolean(False)

498

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

507

@property

508

def value_to_set(self):

509

return self._vts

510

@value_to_set.setter

511

def value_to_set(self, value):

512

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

513

self._vts = value.read()

514

value.close()

515

516

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

517

propname = "Timeout"

518

519

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

520

propname = "ExtendedTimeout"

521

522

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

523

propname = "Interval"

524

525

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

526

propname = "ApprovalDelay"

527

528

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

529

propname = "ApprovalDuration"

530

531

def add_command_line_options(parser):

532

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

533

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

534

help="show version number and exit")

535

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

536

help="Select all clients")

537

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

538

help="Print all fields")

539

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

540

help="Dump client data in JSON format")

541

enable_disable = parser.add_mutually_exclusive_group()

542

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

543

help="Enable client")

544

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

545

action="store_true",

546

help="disable client")

547

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

548

help="Bump timeout for client")

549

start_stop_checker = parser.add_mutually_exclusive_group()

550

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

551

action="store_true",

552

help="Start checker for client")

553

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

554

action="store_true",

555

help="Stop checker for client")

556

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

557

help="Check if client is enabled")

558

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

559

help="Remove client")

560

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

561

help="Set checker command for client")

562

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

563

help="Set timeout for client")

564

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

565

help="Set extended timeout for client")

566

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

567

help="Set checker interval for client")

568

approve_deny_default = parser.add_mutually_exclusive_group()

569

approve_deny_default.add_argument(

570

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

571

default=None, dest="approved_by_default",

572

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

573

approve_deny_default.add_argument(

574

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

575

dest="approved_by_default",

576

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

577

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

578

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

579

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

580

help="Set duration of one client approval")

581

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

582

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

583

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

584

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

585

approve_deny = parser.add_mutually_exclusive_group()

586

approve_deny.add_argument(

587

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

588

help="Approve any current client request")

589

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

590

help="Deny any current client request")

591

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

592

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

593

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

594

help="Run self-test")

595

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

596

597

598

def commands_from_options(options):

599

600

commands = []

601

602

if options.dump_json:

603

commands.append(DumpJSONCmd())

604

605

if options.enable:

606

commands.append(EnableCmd())

607

608

if options.disable:

609

commands.append(DisableCmd())

610

611

if options.bump_timeout:

612

commands.append(BumpTimeoutCmd())

613

614

if options.start_checker:

615

commands.append(StartCheckerCmd())

616

617

if options.stop_checker:

618

commands.append(StopCheckerCmd())

619

620

if options.is_enabled:

621

commands.append(IsEnabledCmd())

622

623

if options.checker is not None:

624

commands.append(SetCheckerCmd(options.checker))

625

626

if options.timeout is not None:

627

commands.append(SetTimeoutCmd(options.timeout))

628

629

if options.extended_timeout:

630

commands.append(

631

SetExtendedTimeoutCmd(options.extended_timeout))

632

633

if options.interval is not None:

634

commands.append(SetIntervalCmd(options.interval))

635

636

if options.approved_by_default is not None:

637

if options.approved_by_default:

638

commands.append(ApproveByDefaultCmd())

639

else:

640

commands.append(DenyByDefaultCmd())

641

642

if options.approval_delay is not None:

643

commands.append(SetApprovalDelayCmd(options.approval_delay))

644

645

if options.approval_duration is not None:

646

commands.append(

647

SetApprovalDurationCmd(options.approval_duration))

648

649

if options.host is not None:

650

commands.append(SetHostCmd(options.host))

651

652

if options.secret is not None:

653

commands.append(SetSecretCmd(options.secret))

654

655

if options.approve:

656

commands.append(ApproveCmd())

657

658

if options.deny:

659

commands.append(DenyCmd())

660

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

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

665

# optionally verbosely

666

if not commands:

667

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

668

669

return commands

670

671

672

def check_option_syntax(parser, options):

673

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

674

argparse"""

675

676

def has_actions(options):

677

return any((options.enable,

678

options.disable,

679

options.bump_timeout,

680

options.start_checker,

681

options.stop_checker,

682

options.is_enabled,

683

options.remove,

684

options.checker is not None,

685

options.timeout is not None,

686

options.extended_timeout is not None,

687

options.interval is not None,

688

options.approved_by_default is not None,

689

options.approval_delay is not None,

690

options.approval_duration is not None,

691

options.host is not None,

692

options.secret is not None,

693

options.approve,

694

options.deny))

695

696

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

697

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

698

if options.verbose and has_actions(options):

699

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

700

if options.dump_json and (options.verbose

701

or has_actions(options)):

702

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

703

if options.all and not has_actions(options):

704

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

705

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

706

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

712

713

714

def main():

715

parser = argparse.ArgumentParser()

716

717

add_command_line_options(parser)

718

719

options = parser.parse_args()

720

721

check_option_syntax(parser, options)

722

723

clientnames = options.client

724

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

try:

729

bus = dbus.SystemBus()

730

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

731

dbus_busname, server_dbus_path)

732

mandos_dbus_objc = bus.get_object(dbus_busname,

733

server_dbus_path)

734

except dbus.exceptions.DBusException:

735

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

736

sys.exit(1)

737

738

mandos_serv = dbus.Interface(mandos_dbus_objc,

739

dbus_interface=server_dbus_interface)

740

mandos_serv_object_manager = dbus.Interface(

741

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

742

743

# Filter out log message from dbus module

744

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

745

class NullFilter(logging.Filter):

746

def filter(self, record):

747

return False

748

dbus_filter = NullFilter()

749

try:

750

dbus_logger.addFilter(dbus_filter)

751

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

752

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

753

mandos_clients = {path: ifs_and_props[client_dbus_interface]

754

for path, ifs_and_props in

755

mandos_serv_object_manager

756

.GetManagedObjects().items()

757

if client_dbus_interface in ifs_and_props}

758

except dbus.exceptions.DBusException as e:

759

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

760

"\n%s", e)

761

sys.exit(1)

762

finally:

763

# restore dbus logger

764

dbus_logger.removeFilter(dbus_filter)

765

766

# Compile dict of (clients: properties) to process

767

clients = {}

768

769

if not clientnames:

770

clients = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

772

else:

773

for name in clientnames:

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

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(clients, bus, mandos_serv)

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__ = "/clients/{}".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_dbus_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_dbus_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,

856

client_dbus_interface)

857

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

858

dbus_interface)))

859

self.client = MockClient(

860

"foo",

861

KeyID=("92ed150794387c03ce684574b1139a65"

862

"94a34f895daaaf09fd8ea90a27cddb12"),

863

Secret=b"secret",

864

Host="foo.example.org",

865

Enabled=dbus.Boolean(True),

866

Timeout=300000,

867

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

868

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

869

Interval=120000,

870

Fingerprint=("778827225BA7DE539C5A"

871

"7CFA59CFF7CDBD9A5920"),

872

CheckerRunning=dbus.Boolean(False),

873

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

874

ApprovalPending=dbus.Boolean(False),

875

ApprovedByDefault=dbus.Boolean(True),

876

LastApprovalRequest="",

877

ApprovalDelay=0,

878

ApprovalDuration=1000,

879

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

880

ExtendedTimeout=900000,

881

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

882

LastCheckerStatus=0)

883

self.other_client = MockClient(

884

"barbar",

885

KeyID=("0558568eedd67d622f5c83b35a115f79"

886

"6ab612cff5ad227247e46c2b020f441c"),

887

Secret=b"secretbar",

888

Host="192.0.2.3",

889

Enabled=dbus.Boolean(True),

890

Timeout=300000,

891

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

892

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

893

Interval=120000,

894

Fingerprint=("3E393AEAEFB84C7E89E2"

895

"F547B3A107558FCA3A27"),

896

CheckerRunning=dbus.Boolean(True),

897

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

898

ApprovalPending=dbus.Boolean(False),

899

ApprovedByDefault=dbus.Boolean(False),

900

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

901

ApprovalDelay=30000,

902

ApprovalDuration=1000,

903

Checker=":",

904

ExtendedTimeout=900000,

905

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

906

LastCheckerStatus=-2)

907

self.clients = collections.OrderedDict(

908

[

909

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

910

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

911

])

912

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

913

@property

914

def bus(self):

915

class Bus(object):

916

@staticmethod

917

def get_object(client_bus_name, path):

918

self.assertEqual(client_bus_name, dbus_busname)

919

return {

920

"/clients/foo": self.client,

921

"/clients/barbar": self.other_client,

922

}[path]

923

return Bus()

924

925

class TestPrintTableCmd(TestCmd):

926

def test_normal(self):

927

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

928

expected_output = "\n".join((

929

"Name Enabled Timeout Last Successful Check",

930

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

931

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

932

))

933

self.assertEqual(output, expected_output)

934

def test_verbose(self):

935

output = PrintTableCmd(verbose=True).output(

936

self.clients.values())

937

expected_output = "\n".join((

938

# First line (headers)

939

"Name Enabled Timeout Last Successful Check Created "

940

" Interval Host Key ID "

941

" Fingerprint "

942

" Check Is Running Last Enabl"

943

"ed Approval Is Pending Approved By Default Last A"

944

"pproval Request Approval Delay Approval Duration Checker"

945

" Extended Timeout Expires Last "

946

"Checker Status",

947

# Second line (client "foo")

948

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

949

"T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684"

950

"574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7"

951

"DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03"

952

"T00:00:00 No Yes "

953

" 00:00:00 00:00:01 fping -"

954

"q -- %(host)s 00:15:00 2019-02-04T00:00:00 0 "

955

" ",

956

# Third line (client "barbar")

957

"barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03"

958

"T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c8"

959

"3b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB8"

960

"4C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04"

961

"T00:00:00 No No 2019-0"

962

"1-03T00:00:00 00:00:30 00:00:01 : "

963

" 00:15:00 2019-02-05T00:00:00 -2 "

964

" ",

965

))

966

self.assertEqual(output, expected_output)

967

def test_one_client(self):

968

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

969

expected_output = """

970

Name Enabled Timeout Last Successful Check

971

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

972

"""[1:-1]

973

self.assertEqual(output, expected_output)

974

975

class TestDumpJSONCmd(TestCmd):

976

def setUp(self):

977

self.expected_json = {

978

"foo": {

979

"Name": "foo",

980

"KeyID": ("92ed150794387c03ce684574b1139a65"

981

"94a34f895daaaf09fd8ea90a27cddb12"),

982

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

983

"Enabled": True,

984

"Timeout": 300000,

985

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

986

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

987

"Interval": 120000,

988

"Fingerprint": ("778827225BA7DE539C5A"

989

"7CFA59CFF7CDBD9A5920"),

990

"CheckerRunning": False,

991

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

992

"ApprovalPending": False,

993

"ApprovedByDefault": True,

994

"LastApprovalRequest": "",

995

"ApprovalDelay": 0,

996

"ApprovalDuration": 1000,

997

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

998

"ExtendedTimeout": 900000,

999

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

1000

"LastCheckerStatus": 0,

1001

},

1002

"barbar": {

1003

"Name": "barbar",

1004

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1005

"6ab612cff5ad227247e46c2b020f441c"),

1006

"Host": "192.0.2.3",

1007

"Enabled": True,

1008

"Timeout": 300000,

1009

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

1010

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

1011

"Interval": 120000,

1012

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1013

"F547B3A107558FCA3A27"),

1014

"CheckerRunning": True,

1015

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

1016

"ApprovalPending": False,

1017

"ApprovedByDefault": False,

1018

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

1019

"ApprovalDelay": 30000,

1020

"ApprovalDuration": 1000,

1021

"Checker": ":",

1022

"ExtendedTimeout": 900000,

1023

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

1024

"LastCheckerStatus": -2,

1025

},

1026

}

1027

return super(TestDumpJSONCmd, self).setUp()

1028

def test_normal(self):

1029

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

1030

self.assertDictEqual(json_data, self.expected_json)

1031

def test_one_client(self):

1032

clients = self.one_client

1033

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

1034

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

1035

self.assertDictEqual(json_data, expected_json)

1036

1037

class TestIsEnabledCmd(TestCmd):

1038

def test_is_enabled(self):

1039

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

1040

properties)

1041

for client, properties

1042

in self.clients.items()))

1043

def test_is_enabled_run_exits_successfully(self):

1044

with self.assertRaises(SystemExit) as e:

1045

IsEnabledCmd().run(self.one_client)

1046

if e.exception.code is not None:

1047

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

1048

else:

1049

self.assertIsNone(e.exception.code)

1050

def test_is_enabled_run_exits_with_failure(self):

1051

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

1052

with self.assertRaises(SystemExit) as e:

1053

IsEnabledCmd().run(self.one_client)

1054

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

1055

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

1056

else:

1057

self.assertIsNotNone(e.exception.code)

1058

1059

class TestRemoveCmd(TestCmd):

1060

def test_remove(self):

1061

class MockMandos(object):

1062

def __init__(self):

1063

self.calls = []

1064

def RemoveClient(self, dbus_path):

1065

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

1066

mandos = MockMandos()

1067

super(TestRemoveCmd, self).setUp()

1068

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

1069

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

1070

for clientpath in self.clients:

1071

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

1072

mandos.calls)

1073

1074

class TestApproveCmd(TestCmd):

1075

def test_approve(self):

1076

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

1077

for clientpath in self.clients:

1078

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

1079

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

1080

client.calls)

1081

1082

class TestDenyCmd(TestCmd):

1083

def test_deny(self):

1084

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

1085

for clientpath in self.clients:

1086

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

1087

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

1088

client.calls)

1089

1090

class TestEnableCmd(TestCmd):

1091

def test_enable(self):

1092

for clientpath in self.clients:

1093

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

1094

client.attributes["Enabled"] = False

1095

1096

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

1097

1098

for clientpath in self.clients:

1099

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

1100

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

1101

1102

class TestDisableCmd(TestCmd):

1103

def test_disable(self):

1104

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

1105

for clientpath in self.clients:

1106

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

1107

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

1108

1109

class Unique(object):

1110

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

1111

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

1112

1113

class TestPropertyCmd(TestCmd):

1114

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

1115

def runTest(self):

1116

if not hasattr(self, "command"):

1117

return

1118

values_to_get = getattr(self, "values_to_get",

1119

self.values_to_set)

1120

for value_to_set, value_to_get in zip(self.values_to_set,

1121

values_to_get):

1122

for clientpath in self.clients:

1123

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

1124

old_value = client.attributes[self.propname]

1125

self.assertNotIsInstance(old_value, Unique)

1126

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

1127

self.run_command(value_to_set, self.clients)

1128

for clientpath in self.clients:

1129

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

1130

value = client.attributes[self.propname]

1131

self.assertNotIsInstance(value, Unique)

1132

self.assertEqual(value, value_to_get)

1133

def run_command(self, value, clients):

1134

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

1135

1136

class TestBumpTimeoutCmd(TestPropertyCmd):

1137

command = BumpTimeoutCmd

1138

propname = "LastCheckedOK"

1139

values_to_set = [""]

1140

1141

class TestStartCheckerCmd(TestPropertyCmd):

1142

command = StartCheckerCmd

1143

propname = "CheckerRunning"

1144

values_to_set = [dbus.Boolean(True)]

1145

1146

class TestStopCheckerCmd(TestPropertyCmd):

1147

command = StopCheckerCmd

1148

propname = "CheckerRunning"

1149

values_to_set = [dbus.Boolean(False)]

1150

1151

class TestApproveByDefaultCmd(TestPropertyCmd):

1152

command = ApproveByDefaultCmd

1153

propname = "ApprovedByDefault"

1154

values_to_set = [dbus.Boolean(True)]

1155

1156

class TestDenyByDefaultCmd(TestPropertyCmd):

1157

command = DenyByDefaultCmd

1158

propname = "ApprovedByDefault"

1159

values_to_set = [dbus.Boolean(False)]

1160

1161

class TestPropertyValueCmd(TestPropertyCmd):

1162

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

1163

def runTest(self):

1164

if type(self) is TestPropertyValueCmd:

1165

return

1166

return super(TestPropertyValueCmd, self).runTest()

1167

def run_command(self, value, clients):

1168

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

1169

1170

class TestSetCheckerCmd(TestPropertyValueCmd):

1171

command = SetCheckerCmd

1172

propname = "Checker"

1173

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

1174

1175

class TestSetHostCmd(TestPropertyValueCmd):

1176

command = SetHostCmd

1177

propname = "Host"

1178

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

1179

1180

class TestSetSecretCmd(TestPropertyValueCmd):

1181

command = SetSecretCmd

1182

propname = "Secret"

1183

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

1184

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

1185

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

1186

1187

class TestSetTimeoutCmd(TestPropertyValueCmd):

1188

command = SetTimeoutCmd

1189

propname = "Timeout"

1190

values_to_set = [datetime.timedelta(),

1191

datetime.timedelta(minutes=5),

1192

datetime.timedelta(seconds=1),

1193

datetime.timedelta(weeks=1),

1194

datetime.timedelta(weeks=52)]

1195

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

1196

1197

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1198

command = SetExtendedTimeoutCmd

1199

propname = "ExtendedTimeout"

1200

values_to_set = [datetime.timedelta(),

1201

datetime.timedelta(minutes=5),

1202

datetime.timedelta(seconds=1),

1203

datetime.timedelta(weeks=1),

1204

datetime.timedelta(weeks=52)]

1205

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

1206

1207

class TestSetIntervalCmd(TestPropertyValueCmd):

1208

command = SetIntervalCmd

1209

propname = "Interval"

1210

values_to_set = [datetime.timedelta(),

1211

datetime.timedelta(minutes=5),

1212

datetime.timedelta(seconds=1),

1213

datetime.timedelta(weeks=1),

1214

datetime.timedelta(weeks=52)]

1215

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

1216

1217

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1218

command = SetApprovalDelayCmd

1219

propname = "ApprovalDelay"

1220

values_to_set = [datetime.timedelta(),

1221

datetime.timedelta(minutes=5),

1222

datetime.timedelta(seconds=1),

1223

datetime.timedelta(weeks=1),

1224

datetime.timedelta(weeks=52)]

1225

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

1226

1227

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1228

command = SetApprovalDurationCmd

1229

propname = "ApprovalDuration"

1230

values_to_set = [datetime.timedelta(),

1231

datetime.timedelta(minutes=5),

1232

datetime.timedelta(seconds=1),

1233

datetime.timedelta(weeks=1),

1234

datetime.timedelta(weeks=52)]

1235

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

1236

1237

class Test_command_from_options(unittest.TestCase):

1238

def setUp(self):

1239

self.parser = argparse.ArgumentParser()

1240

add_command_line_options(self.parser)

1241

def assert_command_from_args(self, args, command_cls,

1242

**cmd_attrs):

1243

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

1244

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

1245

options = self.parser.parse_args(args)

1246

check_option_syntax(self.parser, options)

1247

commands = commands_from_options(options)

1248

self.assertEqual(len(commands), 1)

1249

command = commands[0]

1250

self.assertIsInstance(command, command_cls)

1251

for key, value in cmd_attrs.items():

1252

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

1253

def test_print_table(self):

1254

self.assert_command_from_args([], PrintTableCmd,

1255

verbose=False)

1256

1257

def test_print_table_verbose(self):

1258

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

1259

verbose=True)

1260

1261

def test_print_table_verbose_short(self):

1262

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

1263

verbose=True)

1264

1265

def test_enable(self):

1266

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

1267

1268

def test_enable_short(self):

1269

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

1270

1271

def test_disable(self):

1272

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

1273

DisableCmd)

1274

1275

def test_disable_short(self):

1276

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

1277

1278

def test_bump_timeout(self):

1279

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

1280

BumpTimeoutCmd)

1281

1282

def test_bump_timeout_short(self):

1283

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

1284

1285

def test_start_checker(self):

1286

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

1287

StartCheckerCmd)

1288

1289

def test_stop_checker(self):

1290

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

1291

StopCheckerCmd)

1292

1293

def test_remove(self):

1294

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

1295

RemoveCmd)

1296

1297

def test_remove_short(self):

1298

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

1299

1300

def test_checker(self):

1301

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

1302

SetCheckerCmd, value_to_set=":")

1303

1304

def test_checker_empty(self):

1305

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

1306

SetCheckerCmd, value_to_set="")

1307

1308

def test_checker_short(self):

1309

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

1310

SetCheckerCmd, value_to_set=":")

1311

1312

def test_timeout(self):

1313

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

1314

SetTimeoutCmd,

1315

value_to_set=300000)

1316

1317

def test_timeout_short(self):

1318

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

1319

SetTimeoutCmd,

1320

value_to_set=300000)

1321

1322

def test_extended_timeout(self):

1323

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

1324

"foo"],

1325

SetExtendedTimeoutCmd,

1326

value_to_set=900000)

1327

1328

def test_interval(self):

1329

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

1330

SetIntervalCmd,

1331

value_to_set=120000)

1332

1333

def test_interval_short(self):

1334

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

1335

SetIntervalCmd,

1336

value_to_set=120000)

1337

1338

def test_approve_by_default(self):

1339

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

1340

ApproveByDefaultCmd)

1341

1342

def test_deny_by_default(self):

1343

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

1344

DenyByDefaultCmd)

1345

1346

def test_approval_delay(self):

1347

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

1348

"foo"], SetApprovalDelayCmd,

1349

value_to_set=30000)

1350

1351

def test_approval_duration(self):

1352

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

1353

"foo"], SetApprovalDurationCmd,

1354

value_to_set=1000)

1355

1356

def test_host(self):

1357

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

1358

"foo"], SetHostCmd,

1359

value_to_set="foo.example.org")

1360

1361

def test_host_short(self):

1362

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

1363

"foo"], SetHostCmd,

1364

value_to_set="foo.example.org")

1365

1366

def test_secret_devnull(self):

1367

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

1368

"foo"], SetSecretCmd,

1369

value_to_set=b"")

1370

1371

def test_secret_tempfile(self):

1372

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

1373

value = b"secret\0xyzzy\nbar"

1374

f.write(value)

1375

f.seek(0)

1376

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

1377

"foo"], SetSecretCmd,

1378

value_to_set=value)

1379

1380

def test_secret_devnull_short(self):

1381

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

1382

SetSecretCmd, value_to_set=b"")

1383

1384

def test_secret_tempfile_short(self):

1385

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

1386

value = b"secret\0xyzzy\nbar"

1387

f.write(value)

1388

f.seek(0)

1389

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

1390

SetSecretCmd,

1391

value_to_set=value)

1392

1393

def test_approve(self):

1394

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

1395

ApproveCmd)

1396

1397

def test_approve_short(self):

1398

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

1399

1400

def test_deny(self):

1401

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

1402

1403

def test_deny_short(self):

1404

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

1405

1406

def test_dump_json(self):

1407

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

1408

1409

def test_is_enabled(self):

1410

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

1411

IsEnabledCmd)

1412

1413

def test_is_enabled_short(self):

1414

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

1415

1416

def test_deny_before_remove(self):

1417

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

1418

"foo"])

1419

check_option_syntax(self.parser, options)

1420

commands = commands_from_options(options)

1421

self.assertEqual(len(commands), 2)

1422

self.assertIsInstance(commands[0], DenyCmd)

1423

self.assertIsInstance(commands[1], RemoveCmd)

1424

1425

def test_deny_before_remove_reversed(self):

1426

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

1427

"--all"])

1428

check_option_syntax(self.parser, options)

1429

commands = commands_from_options(options)

1430

self.assertEqual(len(commands), 2)

1431

self.assertIsInstance(commands[0], DenyCmd)

1432

self.assertIsInstance(commands[1], RemoveCmd)

1433

1434

1435

class Test_check_option_syntax(unittest.TestCase):

1436

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

1437

# check_option_syntax()

1438

actions = {

1439

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

1440

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

1441

# never used

1442

"enable": True,

1443

"disable": True,

1444

"bump_timeout": True,

1445

"start_checker": True,

1446

"stop_checker": True,

1447

"is_enabled": True,

1448

"remove": True,

1449

"checker": "x",

1450

"timeout": datetime.timedelta(),

1451

"extended_timeout": datetime.timedelta(),

1452

"interval": datetime.timedelta(),

1453

"approved_by_default": True,

1454

"approval_delay": datetime.timedelta(),

1455

"approval_duration": datetime.timedelta(),

1456

"host": "x",

1457

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

1458

"approve": True,

1459

"deny": True,

1460

}

1461

1462

def setUp(self):

1463

self.parser = argparse.ArgumentParser()

1464

add_command_line_options(self.parser)

1465

1466

@contextlib.contextmanager

1467

def assertParseError(self):

1468

with self.assertRaises(SystemExit) as e:

1469

with self.temporarily_suppress_stderr():

1470

yield

1471

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

1472

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

1473

# /argparse.html#exiting-methods

1474

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

1475

1476

@staticmethod

1477

@contextlib.contextmanager

1478

def temporarily_suppress_stderr():

1479

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

1480

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

1481

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

1482

os.close(null)

1483

try:

1484

yield

1485

finally:

1486

# restore stderr

1487

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

1488

os.close(stderrcopy)

1489

1490

def check_option_syntax(self, options):

1491

check_option_syntax(self.parser, options)

1492

1493

def test_actions_requires_client_or_all(self):

1494

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

1495

options = self.parser.parse_args()

1496

setattr(options, action, value)

1497

with self.assertParseError():

1498

self.check_option_syntax(options)

1499

1500

def test_actions_conflicts_with_verbose(self):

1501

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

1502

options = self.parser.parse_args()

1503

setattr(options, action, value)

1504

options.verbose = True

1505

with self.assertParseError():

1506

self.check_option_syntax(options)

1507

1508

def test_dump_json_conflicts_with_verbose(self):

1509

options = self.parser.parse_args()

1510

options.dump_json = True

1511

options.verbose = True

1512

with self.assertParseError():

1513

self.check_option_syntax(options)

1514

1515

def test_dump_json_conflicts_with_action(self):

1516

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

1517

options = self.parser.parse_args()

1518

setattr(options, action, value)

1519

options.dump_json = True

1520

with self.assertParseError():

1521

self.check_option_syntax(options)

1522

1523

def test_all_can_not_be_alone(self):

1524

options = self.parser.parse_args()

1525

options.all = True

1526

with self.assertParseError():

1527

self.check_option_syntax(options)

1528

1529

def test_all_is_ok_with_any_action(self):

1530

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

1531

options = self.parser.parse_args()

1532

setattr(options, action, value)

1533

options.all = True

1534

self.check_option_syntax(options)

1535

1536

def test_is_enabled_fails_without_client(self):

1537

options = self.parser.parse_args()

1538

options.is_enabled = True

1539

with self.assertParseError():

1540

self.check_option_syntax(options)

1541

1542

def test_is_enabled_works_with_one_client(self):

1543

options = self.parser.parse_args()

1544

options.is_enabled = True

1545

options.client = ["foo"]

1546

self.check_option_syntax(options)

1547

1548

def test_is_enabled_fails_with_two_clients(self):

1549

options = self.parser.parse_args()

1550

options.is_enabled = True

1551

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

1552

with self.assertParseError():

1553

self.check_option_syntax(options)

1554

1555

def test_remove_can_only_be_combined_with_action_deny(self):

1556

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

1557

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

1558

continue

1559

options = self.parser.parse_args()

1560

setattr(options, action, value)

1561

options.all = True

1562

options.remove = True

1563

with self.assertParseError():

1564

self.check_option_syntax(options)

1565

1566

1567

1568

def should_only_run_tests():

1569

parser = argparse.ArgumentParser(add_help=False)

1570

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

1571

args, unknown_args = parser.parse_known_args()

1572

run_tests = args.check

1573

if run_tests:

1574

# Remove --check argument from sys.argv

1575

sys.argv[1:] = unknown_args

1576

return run_tests

1577

1578

# Add all tests from doctest strings

1579

def load_tests(loader, tests, none):

1580

import doctest

1581

tests.addTests(doctest.DocTestSuite())

1582

return tests

1583

1584

if __name__ == "__main__":

1585

if should_only_run_tests():

1586

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

1587

unittest.main()

1588

else:

1589

main()