/mandos/release : revision 237.26.2

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, mandos_objmgr, 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, mandos_objmgr, clients):

297

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

298

299

class PropertyCmd(Command):

300

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

301

def run_on_one_client(self, client, properties):

302

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

303

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

304

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

307

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

308

else bool(self.value_to_set))

309

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

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

312

class ValueArgumentMixIn(object):

313

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

314

def __init__(self, value):

315

self.value_to_set = value

316

317

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

318

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

319

milliseconds."""

320

@property

321

def value_to_set(self):

322

return self._vts

323

@value_to_set.setter

324

def value_to_set(self, value):

325

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

326

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

327

328

# Actual (non-abstract) command classes

329

330

class PrintTableCmd(PrintCmd):

331

def __init__(self, verbose=False):

332

self.verbose = verbose

333

334

def output(self, mandos_objmgr, clients):

335

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

336

keywords = default_keywords

337

if self.verbose:

338

keywords = self.all_keywords

339

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

340

341

class TableOfClients(object):

342

tableheaders = {

343

"Name": "Name",

344

"Enabled": "Enabled",

345

"Timeout": "Timeout",

346

"LastCheckedOK": "Last Successful Check",

347

"LastApprovalRequest": "Last Approval Request",

348

"Created": "Created",

349

"Interval": "Interval",

350

"Host": "Host",

351

"Fingerprint": "Fingerprint",

352

"KeyID": "Key ID",

353

"CheckerRunning": "Check Is Running",

354

"LastEnabled": "Last Enabled",

355

"ApprovalPending": "Approval Is Pending",

356

"ApprovedByDefault": "Approved By Default",

357

"ApprovalDelay": "Approval Delay",

358

"ApprovalDuration": "Approval Duration",

359

"Checker": "Checker",

360

"ExtendedTimeout": "Extended Timeout",

361

"Expires": "Expires",

362

"LastCheckerStatus": "Last Checker Status",

363

}

364

365

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

366

self.clients = clients

367

self.keywords = keywords

368

if tableheaders is not None:

369

self.tableheaders = tableheaders

370

371

def __str__(self):

372

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

373

374

if sys.version_info.major == 2:

375

__unicode__ = __str__

376

def __str__(self):

377

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

378

379

def rows(self):

380

format_string = self.row_formatting_string()

381

rows = [self.header_line(format_string)]

382

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

383

for client in self.clients)

384

return rows

385

386

def row_formatting_string(self):

387

"Format string used to format table rows"

388

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

389

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

390

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

391

for client in self.clients)),

392

key=key)

393

for key in self.keywords)

394

395

def string_from_client(self, client, key):

396

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

397

398

@staticmethod

399

def valuetostring(value, keyword):

400

if isinstance(value, dbus.Boolean):

401

return "Yes" if value else "No"

402

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

403

"ApprovalDuration", "ExtendedTimeout"):

404

return milliseconds_to_string(value)

405

return str(value)

406

407

def header_line(self, format_string):

408

return format_string.format(**self.tableheaders)

409

410

def client_line(self, client, format_string):

411

return format_string.format(

412

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

413

for key in self.keywords})

414

415

416

417

class DumpJSONCmd(PrintCmd):

418

def output(self, mandos_objmgr, clients):

419

data = {client["Name"]:

420

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

421

for key in self.all_keywords}

422

for client in clients.values()}

423

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

424

@staticmethod

425

def dbus_boolean_to_bool(value):

426

if isinstance(value, dbus.Boolean):

427

value = bool(value)

428

return value

429

430

class IsEnabledCmd(Command):

431

def run_on_one_client(self, client, properties):

432

if self.is_enabled(client, properties):

433

sys.exit(0)

434

sys.exit(1)

435

def is_enabled(self, client, properties):

436

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

437

client.__dbus_object_path__,

438

dbus.PROPERTIES_IFACE, client_interface,

439

"Enabled")

440

return bool(client.Get(client_interface, "Enabled",

441

dbus_interface=dbus.PROPERTIES_IFACE))

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

property = "Enabled"

466

value_to_set = dbus.Boolean(True)

467

468

class DisableCmd(PropertyCmd):

469

property = "Enabled"

470

value_to_set = dbus.Boolean(False)

471

472

class BumpTimeoutCmd(PropertyCmd):

473

property = "LastCheckedOK"

474

value_to_set = ""

475

476

class StartCheckerCmd(PropertyCmd):

477

property = "CheckerRunning"

478

value_to_set = dbus.Boolean(True)

479

480

class StopCheckerCmd(PropertyCmd):

481

property = "CheckerRunning"

482

value_to_set = dbus.Boolean(False)

483

484

class ApproveByDefaultCmd(PropertyCmd):

485

property = "ApprovedByDefault"

486

value_to_set = dbus.Boolean(True)

487

488

class DenyByDefaultCmd(PropertyCmd):

489

property = "ApprovedByDefault"

490

value_to_set = dbus.Boolean(False)

491

492

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

493

property = "Checker"

494

495

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

496

property = "Host"

497

498

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

499

@property

500

def value_to_set(self):

501

return self._vts

502

@value_to_set.setter

503

def value_to_set(self, value):

504

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

505

self._vts = value.read()

506

value.close()

507

property = "Secret"

508

509

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

510

property = "Timeout"

511

512

class SetExtendedTimeoutCmd(PropertyCmd,

513

MillisecondsValueArgumentMixIn):

514

property = "ExtendedTimeout"

515

516

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

517

property = "Interval"

518

519

class SetApprovalDelayCmd(PropertyCmd,

520

MillisecondsValueArgumentMixIn):

521

property = "ApprovalDelay"

522

523

class SetApprovalDurationCmd(PropertyCmd,

524

MillisecondsValueArgumentMixIn):

525

property = "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, mandos_serv_object_manager, 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, property, value, dbus_interface):

845

testcase.assertEqual(interface, client_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[property] = value

849

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

850

testcase.assertEqual(interface, client_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[property]

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(None, 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

None, 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(None, 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(None,

994

self.clients))

995

self.assertDictEqual(json_data, self.expected_json)

996

def test_one_client(self):

997

clients = self.one_client

998

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

999

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

1000

self.assertDictEqual(json_data, expected_json)

1001

1002

class TestIsEnabledCmd(TestCmd):

1003

def test_is_enabled(self):

1004

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

1005

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

1006

def test_is_enabled_run_exits_successfully(self):

1007

with self.assertRaises(SystemExit) as e:

1008

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

1009

if e.exception.code is not None:

1010

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

1011

else:

1012

self.assertIsNone(e.exception.code)

1013

def test_is_enabled_run_exits_with_failure(self):

1014

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

1015

with self.assertRaises(SystemExit) as e:

1016

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

1017

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

1018

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

1019

else:

1020

self.assertIsNotNone(e.exception.code)

1021

1022

class TestRemoveCmd(TestCmd):

1023

def test_remove(self):

1024

class MockMandos(object):

1025

def __init__(self):

1026

self.calls = []

1027

def RemoveClient(self, dbus_path):

1028

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

1029

mandos = MockMandos()

1030

super(TestRemoveCmd, self).setUp()

1031

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

1032

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

1033

for client in self.clients:

1034

self.assertIn(("RemoveClient",

1035

(client.__dbus_object_path__,)),

1036

mandos.calls)

1037

1038

class TestApproveCmd(TestCmd):

1039

def test_approve(self):

1040

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

1041

for client in self.clients:

1042

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

1043

client.calls)

1044

1045

class TestDenyCmd(TestCmd):

1046

def test_deny(self):

1047

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

1048

for client in self.clients:

1049

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

1050

client.calls)

1051

1052

class TestEnableCmd(TestCmd):

1053

def test_enable(self):

1054

for client in self.clients:

1055

client.attributes["Enabled"] = False

1056

1057

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

1058

1059

for client in self.clients:

1060

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

1061

1062

class TestDisableCmd(TestCmd):

1063

def test_disable(self):

1064

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

1065

1066

for client in self.clients:

1067

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

1068

1069

class Unique(object):

1070

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

1071

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

1072

1073

class TestPropertyCmd(TestCmd):

1074

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

1075

def runTest(self):

1076

if not hasattr(self, "command"):

1077

return

1078

values_to_get = getattr(self, "values_to_get",

1079

self.values_to_set)

1080

for value_to_set, value_to_get in zip(self.values_to_set,

1081

values_to_get):

1082

for client in self.clients:

1083

old_value = client.attributes[self.property]

1084

self.assertNotIsInstance(old_value, Unique)

1085

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

1086

self.run_command(value_to_set, self.clients)

1087

for client in self.clients:

1088

value = client.attributes[self.property]

1089

self.assertNotIsInstance(value, Unique)

1090

self.assertEqual(value, value_to_get)

1091

def run_command(self, value, clients):

1092

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

1093

1094

class TestBumpTimeoutCmd(TestPropertyCmd):

1095

command = BumpTimeoutCmd

1096

property = "LastCheckedOK"

1097

values_to_set = [""]

1098

1099

class TestStartCheckerCmd(TestPropertyCmd):

1100

command = StartCheckerCmd

1101

property = "CheckerRunning"

1102

values_to_set = [dbus.Boolean(True)]

1103

1104

class TestStopCheckerCmd(TestPropertyCmd):

1105

command = StopCheckerCmd

1106

property = "CheckerRunning"

1107

values_to_set = [dbus.Boolean(False)]

1108

1109

class TestApproveByDefaultCmd(TestPropertyCmd):

1110

command = ApproveByDefaultCmd

1111

property = "ApprovedByDefault"

1112

values_to_set = [dbus.Boolean(True)]

1113

1114

class TestDenyByDefaultCmd(TestPropertyCmd):

1115

command = DenyByDefaultCmd

1116

property = "ApprovedByDefault"

1117

values_to_set = [dbus.Boolean(False)]

1118

1119

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1120

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

1121

ValueArgumentMixIn"""

1122

def runTest(self):

1123

if type(self) is TestValueArgumentPropertyCmd:

1124

return

1125

return super(TestValueArgumentPropertyCmd, self).runTest()

1126

def run_command(self, value, clients):

1127

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

1128

1129

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1130

command = SetCheckerCmd

1131

property = "Checker"

1132

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

1133

1134

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1135

command = SetHostCmd

1136

property = "Host"

1137

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

1138

1139

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1140

command = SetSecretCmd

1141

property = "Secret"

1142

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

1143

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

1144

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

1145

1146

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1147

command = SetTimeoutCmd

1148

property = "Timeout"

1149

values_to_set = [datetime.timedelta(),

1150

datetime.timedelta(minutes=5),

1151

datetime.timedelta(seconds=1),

1152

datetime.timedelta(weeks=1),

1153

datetime.timedelta(weeks=52)]

1154

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

1155

1156

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1157

command = SetExtendedTimeoutCmd

1158

property = "ExtendedTimeout"

1159

values_to_set = [datetime.timedelta(),

1160

datetime.timedelta(minutes=5),

1161

datetime.timedelta(seconds=1),

1162

datetime.timedelta(weeks=1),

1163

datetime.timedelta(weeks=52)]

1164

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

1165

1166

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1167

command = SetIntervalCmd

1168

property = "Interval"

1169

values_to_set = [datetime.timedelta(),

1170

datetime.timedelta(minutes=5),

1171

datetime.timedelta(seconds=1),

1172

datetime.timedelta(weeks=1),

1173

datetime.timedelta(weeks=52)]

1174

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

1175

1176

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1177

command = SetApprovalDelayCmd

1178

property = "ApprovalDelay"

1179

values_to_set = [datetime.timedelta(),

1180

datetime.timedelta(minutes=5),

1181

datetime.timedelta(seconds=1),

1182

datetime.timedelta(weeks=1),

1183

datetime.timedelta(weeks=52)]

1184

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

1185

1186

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1187

command = SetApprovalDurationCmd

1188

property = "ApprovalDuration"

1189

values_to_set = [datetime.timedelta(),

1190

datetime.timedelta(minutes=5),

1191

datetime.timedelta(seconds=1),

1192

datetime.timedelta(weeks=1),

1193

datetime.timedelta(weeks=52)]

1194

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

1195

1196

class Test_command_from_options(unittest.TestCase):

1197

def setUp(self):

1198

self.parser = argparse.ArgumentParser()

1199

add_command_line_options(self.parser)

1200

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

1201

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

1202

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

1203

options = self.parser.parse_args(args)

1204

check_option_syntax(self.parser, options)

1205

commands = commands_from_options(options)

1206

self.assertEqual(len(commands), 1)

1207

command = commands[0]

1208

self.assertIsInstance(command, command_cls)

1209

for key, value in cmd_attrs.items():

1210

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

1211

def test_print_table(self):

1212

self.assert_command_from_args([], PrintTableCmd,

1213

verbose=False)

1214

1215

def test_print_table_verbose(self):

1216

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

1217

verbose=True)

1218

1219

def test_print_table_verbose_short(self):

1220

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

1221

verbose=True)

1222

1223

def test_enable(self):

1224

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

1225

1226

def test_enable_short(self):

1227

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

1228

1229

def test_disable(self):

1230

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

1231

DisableCmd)

1232

1233

def test_disable_short(self):

1234

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

1235

1236

def test_bump_timeout(self):

1237

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

1238

BumpTimeoutCmd)

1239

1240

def test_bump_timeout_short(self):

1241

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

1242

1243

def test_start_checker(self):

1244

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

1245

StartCheckerCmd)

1246

1247

def test_stop_checker(self):

1248

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

1249

StopCheckerCmd)

1250

1251

def test_remove(self):

1252

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

1253

RemoveCmd)

1254

1255

def test_remove_short(self):

1256

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

1257

1258

def test_checker(self):

1259

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

1260

SetCheckerCmd, value_to_set=":")

1261

1262

def test_checker_empty(self):

1263

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

1264

SetCheckerCmd, value_to_set="")

1265

1266

def test_checker_short(self):

1267

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

1268

SetCheckerCmd, value_to_set=":")

1269

1270

def test_timeout(self):

1271

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

1272

SetTimeoutCmd,

1273

value_to_set=300000)

1274

1275

def test_timeout_short(self):

1276

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

1277

SetTimeoutCmd,

1278

value_to_set=300000)

1279

1280

def test_extended_timeout(self):

1281

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

1282

"foo"],

1283

SetExtendedTimeoutCmd,

1284

value_to_set=900000)

1285

1286

def test_interval(self):

1287

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

1288

SetIntervalCmd,

1289

value_to_set=120000)

1290

1291

def test_interval_short(self):

1292

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

1293

SetIntervalCmd,

1294

value_to_set=120000)

1295

1296

def test_approve_by_default(self):

1297

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

1298

ApproveByDefaultCmd)

1299

1300

def test_deny_by_default(self):

1301

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

1302

DenyByDefaultCmd)

1303

1304

def test_approval_delay(self):

1305

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

1306

"foo"], SetApprovalDelayCmd,

1307

value_to_set=30000)

1308

1309

def test_approval_duration(self):

1310

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

1311

"foo"], SetApprovalDurationCmd,

1312

value_to_set=1000)

1313

1314

def test_host(self):

1315

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

1316

"foo"], SetHostCmd,

1317

value_to_set="foo.example.org")

1318

1319

def test_host_short(self):

1320

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

1321

"foo"], SetHostCmd,

1322

value_to_set="foo.example.org")

1323

1324

def test_secret_devnull(self):

1325

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

1326

"foo"], SetSecretCmd,

1327

value_to_set=b"")

1328

1329

def test_secret_tempfile(self):

1330

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

1331

value = b"secret\0xyzzy\nbar"

1332

f.write(value)

1333

f.seek(0)

1334

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

1335

"foo"], SetSecretCmd,

1336

value_to_set=value)

1337

1338

def test_secret_devnull_short(self):

1339

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

1340

SetSecretCmd, value_to_set=b"")

1341

1342

def test_secret_tempfile_short(self):

1343

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

1344

value = b"secret\0xyzzy\nbar"

1345

f.write(value)

1346

f.seek(0)

1347

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

1348

SetSecretCmd,

1349

value_to_set=value)

1350

1351

def test_approve(self):

1352

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

1353

ApproveCmd)

1354

1355

def test_approve_short(self):

1356

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

1357

1358

def test_deny(self):

1359

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

1360

1361

def test_deny_short(self):

1362

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

1363

1364

def test_dump_json(self):

1365

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

1366

1367

def test_is_enabled(self):

1368

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

1369

IsEnabledCmd)

1370

1371

def test_is_enabled_short(self):

1372

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

1373

1374

def test_deny_before_remove(self):

1375

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

1376

check_option_syntax(self.parser, options)

1377

commands = commands_from_options(options)

1378

self.assertEqual(len(commands), 2)

1379

self.assertIsInstance(commands[0], DenyCmd)

1380

self.assertIsInstance(commands[1], RemoveCmd)

1381

1382

def test_deny_before_remove_reversed(self):

1383

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

1384

check_option_syntax(self.parser, options)

1385

commands = commands_from_options(options)

1386

self.assertEqual(len(commands), 2)

1387

self.assertIsInstance(commands[0], DenyCmd)

1388

self.assertIsInstance(commands[1], RemoveCmd)

1389

1390

1391

class Test_check_option_syntax(unittest.TestCase):

1392

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

1393

# check_option_syntax()

1394

actions = {

1395

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

1396

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

1397

# never used

1398

"enable": True,

1399

"disable": True,

1400

"bump_timeout": True,

1401

"start_checker": True,

1402

"stop_checker": True,

1403

"is_enabled": True,

1404

"remove": True,

1405

"checker": "x",

1406

"timeout": datetime.timedelta(),

1407

"extended_timeout": datetime.timedelta(),

1408

"interval": datetime.timedelta(),

1409

"approved_by_default": True,

1410

"approval_delay": datetime.timedelta(),

1411

"approval_duration": datetime.timedelta(),

1412

"host": "x",

1413

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

1414

"approve": True,

1415

"deny": True,

1416

}

1417

1418

def setUp(self):

1419

self.parser = argparse.ArgumentParser()

1420

add_command_line_options(self.parser)

1421

1422

@contextlib.contextmanager

1423

def assertParseError(self):

1424

with self.assertRaises(SystemExit) as e:

1425

with self.temporarily_suppress_stderr():

1426

yield

1427

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

1428

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

1429

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

1430

1431

@staticmethod

1432

@contextlib.contextmanager

1433

def temporarily_suppress_stderr():

1434

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

1435

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

1436

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

1437

os.close(null)

1438

try:

1439

yield

1440

finally:

1441

# restore stderr

1442

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

1443

os.close(stderrcopy)

1444

1445

def check_option_syntax(self, options):

1446

check_option_syntax(self.parser, options)

1447

1448

def test_actions_requires_client_or_all(self):

1449

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

1450

options = self.parser.parse_args()

1451

setattr(options, action, value)

1452

with self.assertParseError():

1453

self.check_option_syntax(options)

1454

1455

def test_actions_conflicts_with_verbose(self):

1456

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

1457

options = self.parser.parse_args()

1458

setattr(options, action, value)

1459

options.verbose = True

1460

with self.assertParseError():

1461

self.check_option_syntax(options)

1462

1463

def test_dump_json_conflicts_with_verbose(self):

1464

options = self.parser.parse_args()

1465

options.dump_json = True

1466

options.verbose = True

1467

with self.assertParseError():

1468

self.check_option_syntax(options)

1469

1470

def test_dump_json_conflicts_with_action(self):

1471

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

1472

options = self.parser.parse_args()

1473

setattr(options, action, value)

1474

options.dump_json = True

1475

with self.assertParseError():

1476

self.check_option_syntax(options)

1477

1478

def test_all_can_not_be_alone(self):

1479

options = self.parser.parse_args()

1480

options.all = True

1481

with self.assertParseError():

1482

self.check_option_syntax(options)

1483

1484

def test_all_is_ok_with_any_action(self):

1485

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

1486

options = self.parser.parse_args()

1487

setattr(options, action, value)

1488

options.all = True

1489

self.check_option_syntax(options)

1490

1491

def test_is_enabled_fails_without_client(self):

1492

options = self.parser.parse_args()

1493

options.is_enabled = True

1494

with self.assertParseError():

1495

self.check_option_syntax(options)

1496

1497

def test_is_enabled_works_with_one_client(self):

1498

options = self.parser.parse_args()

1499

options.is_enabled = True

1500

options.client = ["foo"]

1501

self.check_option_syntax(options)

1502

1503

def test_is_enabled_fails_with_two_clients(self):

1504

options = self.parser.parse_args()

1505

options.is_enabled = True

1506

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

1507

with self.assertParseError():

1508

self.check_option_syntax(options)

1509

1510

def test_remove_can_only_be_combined_with_action_deny(self):

1511

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

1512

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

1513

continue

1514

options = self.parser.parse_args()

1515

setattr(options, action, value)

1516

options.all = True

1517

options.remove = True

1518

with self.assertParseError():

1519

self.check_option_syntax(options)

1520

1521

1522

1523

def should_only_run_tests():

1524

parser = argparse.ArgumentParser(add_help=False)

1525

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

1526

args, unknown_args = parser.parse_known_args()

1527

run_tests = args.check

1528

if run_tests:

1529

# Remove --check argument from sys.argv

1530

sys.argv[1:] = unknown_args

1531

return run_tests

1532

1533

# Add all tests from doctest strings

1534

def load_tests(loader, tests, none):

1535

import doctest

1536

tests.addTests(doctest.DocTestSuite())

1537

return tests

1538

1539

if __name__ == "__main__":

1540

if should_only_run_tests():

1541

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

1542

unittest.main()

1543

else:

1544

main()