/mandos/release : revision 24.1.73

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", "LastCheckedOK")

346

keywords = default_keywords

347

if self.verbose:

348

keywords = self.all_keywords

349

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

350

351

class TableOfClients(object):

352

tableheaders = {

353

"Name": "Name",

354

"Enabled": "Enabled",

355

"Timeout": "Timeout",

356

"LastCheckedOK": "Last Successful Check",

357

"LastApprovalRequest": "Last Approval Request",

358

"Created": "Created",

359

"Interval": "Interval",

360

"Host": "Host",

361

"Fingerprint": "Fingerprint",

362

"KeyID": "Key ID",

363

"CheckerRunning": "Check Is Running",

364

"LastEnabled": "Last Enabled",

365

"ApprovalPending": "Approval Is Pending",

366

"ApprovedByDefault": "Approved By Default",

367

"ApprovalDelay": "Approval Delay",

368

"ApprovalDuration": "Approval Duration",

369

"Checker": "Checker",

370

"ExtendedTimeout": "Extended Timeout",

371

"Expires": "Expires",

372

"LastCheckerStatus": "Last Checker Status",

373

}

374

375

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

376

self.clients = clients

377

self.keywords = keywords

378

if tableheaders is not None:

379

self.tableheaders = tableheaders

380

381

def __str__(self):

382

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

383

384

if sys.version_info.major == 2:

385

__unicode__ = __str__

386

def __str__(self):

387

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

388

389

def rows(self):

390

format_string = self.row_formatting_string()

391

rows = [self.header_line(format_string)]

392

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

393

for client in self.clients)

394

return rows

395

396

def row_formatting_string(self):

397

"Format string used to format table rows"

398

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

399

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

400

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

401

for client in self.clients)),

402

key=key)

403

for key in self.keywords)

404

405

def string_from_client(self, client, key):

406

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

407

408

@staticmethod

409

def valuetostring(value, keyword):

410

if isinstance(value, dbus.Boolean):

411

return "Yes" if value else "No"

412

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

413

"ApprovalDuration", "ExtendedTimeout"):

414

return milliseconds_to_string(value)

415

return str(value)

416

417

def header_line(self, format_string):

418

return format_string.format(**self.tableheaders)

419

420

def client_line(self, client, format_string):

421

return format_string.format(

422

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

423

for key in self.keywords})

424

425

426

427

class DumpJSONCmd(PrintCmd):

428

def output(self, clients):

429

data = {client["Name"]:

430

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

431

for key in self.all_keywords}

432

for client in clients.values()}

433

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

434

@staticmethod

435

def dbus_boolean_to_bool(value):

436

if isinstance(value, dbus.Boolean):

437

value = bool(value)

438

return value

439

440

class IsEnabledCmd(Command):

441

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

442

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

443

if self.is_enabled(client, properties):

444

sys.exit(0)

445

sys.exit(1)

446

def is_enabled(self, client, properties):

447

return properties["Enabled"]

448

449

class RemoveCmd(Command):

450

def run_on_one_client(self, client, properties):

451

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

452

server_dbus_path, server_dbus_interface,

453

str(client.__dbus_object_path__))

454

self.mandos.RemoveClient(client.__dbus_object_path__)

455

456

class ApproveCmd(Command):

457

def run_on_one_client(self, client, properties):

458

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

459

client.__dbus_object_path__, client_dbus_interface)

460

client.Approve(dbus.Boolean(True),

461

dbus_interface=client_dbus_interface)

462

463

class DenyCmd(Command):

464

def run_on_one_client(self, client, properties):

465

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

466

client.__dbus_object_path__, client_dbus_interface)

467

client.Approve(dbus.Boolean(False),

468

dbus_interface=client_dbus_interface)

469

470

class EnableCmd(PropertyCmd):

471

propname = "Enabled"

472

value_to_set = dbus.Boolean(True)

473

474

class DisableCmd(PropertyCmd):

475

propname = "Enabled"

476

value_to_set = dbus.Boolean(False)

477

478

class BumpTimeoutCmd(PropertyCmd):

479

propname = "LastCheckedOK"

480

value_to_set = ""

481

482

class StartCheckerCmd(PropertyCmd):

483

propname = "CheckerRunning"

484

value_to_set = dbus.Boolean(True)

485

486

class StopCheckerCmd(PropertyCmd):

487

propname = "CheckerRunning"

488

value_to_set = dbus.Boolean(False)

489

490

class ApproveByDefaultCmd(PropertyCmd):

491

propname = "ApprovedByDefault"

492

value_to_set = dbus.Boolean(True)

493

494

class DenyByDefaultCmd(PropertyCmd):

495

propname = "ApprovedByDefault"

496

value_to_set = dbus.Boolean(False)

497

498

class SetCheckerCmd(PropertyValueCmd):

499

propname = "Checker"

500

501

class SetHostCmd(PropertyValueCmd):

502

propname = "Host"

503

504

class SetSecretCmd(PropertyValueCmd):

505

propname = "Secret"

506

@property

507

def value_to_set(self):

508

return self._vts

509

@value_to_set.setter

510

def value_to_set(self, value):

511

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

512

self._vts = value.read()

513

value.close()

514

515

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

516

propname = "Timeout"

517

518

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

519

propname = "ExtendedTimeout"

520

521

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

522

propname = "Interval"

523

524

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

525

propname = "ApprovalDelay"

526

527

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

528

propname = "ApprovalDuration"

529

530

def add_command_line_options(parser):

531

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

532

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

533

help="show version number and exit")

534

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

535

help="Select all clients")

536

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

537

help="Print all fields")

538

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

539

help="Dump client data in JSON format")

540

enable_disable = parser.add_mutually_exclusive_group()

541

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

542

help="Enable client")

543

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

544

action="store_true",

545

help="disable client")

546

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

547

help="Bump timeout for client")

548

start_stop_checker = parser.add_mutually_exclusive_group()

549

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

550

action="store_true",

551

help="Start checker for client")

552

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

553

action="store_true",

554

help="Stop checker for client")

555

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

556

help="Check if client is enabled")

557

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

558

help="Remove client")

559

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

560

help="Set checker command for client")

561

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

562

help="Set timeout for client")

563

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

564

help="Set extended timeout for client")

565

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

566

help="Set checker interval for client")

567

approve_deny_default = parser.add_mutually_exclusive_group()

568

approve_deny_default.add_argument(

569

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

570

default=None, dest="approved_by_default",

571

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

572

approve_deny_default.add_argument(

573

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

574

dest="approved_by_default",

575

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

576

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

577

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

578

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

579

help="Set duration of one client approval")

580

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

581

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

582

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

583

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

584

approve_deny = parser.add_mutually_exclusive_group()

585

approve_deny.add_argument(

586

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

587

help="Approve any current client request")

588

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

589

help="Deny any current client request")

590

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

591

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

592

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

593

help="Run self-test")

594

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

595

596

597

def commands_from_options(options):

598

599

commands = []

600

601

if options.dump_json:

602

commands.append(DumpJSONCmd())

603

604

if options.enable:

605

commands.append(EnableCmd())

606

607

if options.disable:

608

commands.append(DisableCmd())

609

610

if options.bump_timeout:

611

commands.append(BumpTimeoutCmd())

612

613

if options.start_checker:

614

commands.append(StartCheckerCmd())

615

616

if options.stop_checker:

617

commands.append(StopCheckerCmd())

618

619

if options.is_enabled:

620

commands.append(IsEnabledCmd())

621

622

if options.checker is not None:

623

commands.append(SetCheckerCmd(options.checker))

624

625

if options.timeout is not None:

626

commands.append(SetTimeoutCmd(options.timeout))

627

628

if options.extended_timeout:

629

commands.append(

630

SetExtendedTimeoutCmd(options.extended_timeout))

631

632

if options.interval is not None:

633

commands.append(SetIntervalCmd(options.interval))

634

635

if options.approved_by_default is not None:

636

if options.approved_by_default:

637

commands.append(ApproveByDefaultCmd())

638

else:

639

commands.append(DenyByDefaultCmd())

640

641

if options.approval_delay is not None:

642

commands.append(SetApprovalDelayCmd(options.approval_delay))

643

644

if options.approval_duration is not None:

645

commands.append(

646

SetApprovalDurationCmd(options.approval_duration))

647

648

if options.host is not None:

649

commands.append(SetHostCmd(options.host))

650

651

if options.secret is not None:

652

commands.append(SetSecretCmd(options.secret))

653

654

if options.approve:

655

commands.append(ApproveCmd())

656

657

if options.deny:

658

commands.append(DenyCmd())

659

660

if options.remove:

661

commands.append(RemoveCmd())

662

663

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

664

# optionally verbosely

665

if not commands:

666

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

667

668

return commands

669

670

671

def check_option_syntax(parser, options):

672

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

673

argparse"""

674

675

def has_actions(options):

676

return any((options.enable,

677

options.disable,

678

options.bump_timeout,

679

options.start_checker,

680

options.stop_checker,

681

options.is_enabled,

682

options.remove,

683

options.checker is not None,

684

options.timeout is not None,

685

options.extended_timeout is not None,

686

options.interval is not None,

687

options.approved_by_default is not None,

688

options.approval_delay is not None,

689

options.approval_duration is not None,

690

options.host is not None,

691

options.secret is not None,

692

options.approve,

693

options.deny))

694

695

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

696

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

697

if options.verbose and has_actions(options):

698

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

699

if options.dump_json and (options.verbose

700

or has_actions(options)):

701

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

702

if options.all and not has_actions(options):

703

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

704

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

705

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

706

if options.remove:

707

options.remove = False

708

if has_actions(options) and not options.deny:

709

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

710

options.remove = True

711

712

713

def main():

714

parser = argparse.ArgumentParser()

715

716

add_command_line_options(parser)

717

718

options = parser.parse_args()

719

720

check_option_syntax(parser, options)

721

722

clientnames = options.client

723

724

if options.debug:

725

log.setLevel(logging.DEBUG)

726

727

try:

728

bus = dbus.SystemBus()

729

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

730

dbus_busname, server_dbus_path)

731

mandos_dbus_objc = bus.get_object(dbus_busname,

732

server_dbus_path)

733

except dbus.exceptions.DBusException:

734

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

735

sys.exit(1)

736

737

mandos_serv = dbus.Interface(mandos_dbus_objc,

738

dbus_interface=server_dbus_interface)

739

mandos_serv_object_manager = dbus.Interface(

740

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

741

742

# Filter out log message from dbus module

743

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

744

class NullFilter(logging.Filter):

745

def filter(self, record):

746

return False

747

dbus_filter = NullFilter()

748

try:

749

dbus_logger.addFilter(dbus_filter)

750

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

751

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

752

mandos_clients = {path: ifs_and_props[client_dbus_interface]

753

for path, ifs_and_props in

754

mandos_serv_object_manager

755

.GetManagedObjects().items()

756

if client_dbus_interface in ifs_and_props}

757

except dbus.exceptions.DBusException as e:

758

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

759

"\n%s", e)

760

sys.exit(1)

761

finally:

762

# restore dbus logger

763

dbus_logger.removeFilter(dbus_filter)

764

765

# Compile dict of (clients: properties) to process

766

clients = {}

767

768

if not clientnames:

769

clients = {objpath: properties

770

for objpath, properties in mandos_clients.items()}

771

else:

772

for name in clientnames:

773

for objpath, properties in mandos_clients.items():

774

if properties["Name"] == name:

775

clients[objpath] = properties

776

break

777

else:

778

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

779

sys.exit(1)

780

781

# Run all commands on clients

782

commands = commands_from_options(options)

783

for command in commands:

784

command.run(clients, bus, mandos_serv)

785

786

787

class Test_milliseconds_to_string(unittest.TestCase):

788

def test_all(self):

789

self.assertEqual(milliseconds_to_string(93785000),

790

"1T02:03:05")

791

def test_no_days(self):

792

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

793

def test_all_zero(self):

794

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

795

def test_no_fractional_seconds(self):

796

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

797

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

798

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

799

800

class Test_string_to_delta(unittest.TestCase):

801

def test_handles_basic_rfc3339(self):

802

self.assertEqual(string_to_delta("PT0S"),

803

datetime.timedelta())

804

self.assertEqual(string_to_delta("P0D"),

805

datetime.timedelta())

806

self.assertEqual(string_to_delta("PT1S"),

807

datetime.timedelta(0, 1))

808

self.assertEqual(string_to_delta("PT2H"),

809

datetime.timedelta(0, 7200))

810

def test_falls_back_to_pre_1_6_1_with_warning(self):

811

# assertLogs only exists in Python 3.4

812

if hasattr(self, "assertLogs"):

813

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

814

value = string_to_delta("2h")

815

else:

816

class WarningFilter(logging.Filter):

817

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

818

def filter(self, record):

819

is_warning = record.levelno >= logging.WARNING

820

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

821

False)

822

return not is_warning

823

warning_filter = WarningFilter()

824

log.addFilter(warning_filter)

825

try:

826

value = string_to_delta("2h")

827

finally:

828

log.removeFilter(warning_filter)

829

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

830

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

831

832

833

class TestCmd(unittest.TestCase):

834

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

835

def setUp(self):

836

testcase = self

837

class MockClient(object):

838

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

839

self.__dbus_object_path__ = "/clients/{}".format(name)

840

self.attributes = attributes

841

self.attributes["Name"] = name

842

self.calls = []

843

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

844

testcase.assertEqual(interface, client_dbus_interface)

845

testcase.assertEqual(dbus_interface,

846

dbus.PROPERTIES_IFACE)

847

self.attributes[propname] = value

848

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

849

testcase.assertEqual(interface, client_dbus_interface)

850

testcase.assertEqual(dbus_interface,

851

dbus.PROPERTIES_IFACE)

852

return self.attributes[propname]

853

def Approve(self, approve, dbus_interface):

854

testcase.assertEqual(dbus_interface,

855

client_dbus_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

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

909

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

910

])

911

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

912

@property

913

def bus(self):

914

class Bus(object):

915

@staticmethod

916

def get_object(client_bus_name, path):

917

self.assertEqual(client_bus_name, dbus_busname)

918

return {

919

"/clients/foo": self.client,

920

"/clients/barbar": self.other_client,

921

}[path]

922

return Bus()

923

924

class TestPrintTableCmd(TestCmd):

925

def test_normal(self):

926

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

927

expected_output = """

928

Name Enabled Timeout Last Successful Check

929

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

930

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

931

"""[1:-1]

932

self.assertEqual(output, expected_output)

933

def test_verbose(self):

934

output = PrintTableCmd(verbose=True).output(

935

self.clients.values())

936

expected_output = """

937

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

938

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

939

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

940

"""[1:-1]

941

self.assertEqual(output, expected_output)

942

def test_one_client(self):

943

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

944

expected_output = """

945

Name Enabled Timeout Last Successful Check

946

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

947

"""[1:-1]

948

self.assertEqual(output, expected_output)

949

950

class TestDumpJSONCmd(TestCmd):

951

def setUp(self):

952

self.expected_json = {

953

"foo": {

954

"Name": "foo",

955

"KeyID": ("92ed150794387c03ce684574b1139a65"

956

"94a34f895daaaf09fd8ea90a27cddb12"),

957

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

958

"Enabled": True,

959

"Timeout": 300000,

960

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

961

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

962

"Interval": 120000,

963

"Fingerprint": ("778827225BA7DE539C5A"

964

"7CFA59CFF7CDBD9A5920"),

965

"CheckerRunning": False,

966

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

967

"ApprovalPending": False,

968

"ApprovedByDefault": True,

969

"LastApprovalRequest": "",

970

"ApprovalDelay": 0,

971

"ApprovalDuration": 1000,

972

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

973

"ExtendedTimeout": 900000,

974

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

975

"LastCheckerStatus": 0,

976

},

977

"barbar": {

978

"Name": "barbar",

979

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

980

"6ab612cff5ad227247e46c2b020f441c"),

981

"Host": "192.0.2.3",

982

"Enabled": True,

983

"Timeout": 300000,

984

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

985

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

986

"Interval": 120000,

987

"Fingerprint": ("3E393AEAEFB84C7E89E2"

988

"F547B3A107558FCA3A27"),

989

"CheckerRunning": True,

990

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

991

"ApprovalPending": False,

992

"ApprovedByDefault": False,

993

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

994

"ApprovalDelay": 30000,

995

"ApprovalDuration": 1000,

996

"Checker": ":",

997

"ExtendedTimeout": 900000,

998

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

999

"LastCheckerStatus": -2,

1000

},

1001

}

1002

return super(TestDumpJSONCmd, self).setUp()

1003

def test_normal(self):

1004

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

1005

self.assertDictEqual(json_data, self.expected_json)

1006

def test_one_client(self):

1007

clients = self.one_client

1008

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

1009

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

1010

self.assertDictEqual(json_data, expected_json)

1011

1012

class TestIsEnabledCmd(TestCmd):

1013

def test_is_enabled(self):

1014

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

1015

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

1016

def test_is_enabled_run_exits_successfully(self):

1017

with self.assertRaises(SystemExit) as e:

1018

IsEnabledCmd().run(self.one_client)

1019

if e.exception.code is not None:

1020

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

1021

else:

1022

self.assertIsNone(e.exception.code)

1023

def test_is_enabled_run_exits_with_failure(self):

1024

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

1025

with self.assertRaises(SystemExit) as e:

1026

IsEnabledCmd().run(self.one_client)

1027

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

1028

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

1029

else:

1030

self.assertIsNotNone(e.exception.code)

1031

1032

class TestRemoveCmd(TestCmd):

1033

def test_remove(self):

1034

class MockMandos(object):

1035

def __init__(self):

1036

self.calls = []

1037

def RemoveClient(self, dbus_path):

1038

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

1039

mandos = MockMandos()

1040

super(TestRemoveCmd, self).setUp()

1041

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

1042

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

1043

for clientpath in self.clients:

1044

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

1045

mandos.calls)

1046

1047

class TestApproveCmd(TestCmd):

1048

def test_approve(self):

1049

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

1050

for clientpath in self.clients:

1051

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

1052

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

1053

client.calls)

1054

1055

class TestDenyCmd(TestCmd):

1056

def test_deny(self):

1057

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

1058

for clientpath in self.clients:

1059

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

1060

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

1061

client.calls)

1062

1063

class TestEnableCmd(TestCmd):

1064

def test_enable(self):

1065

for clientpath in self.clients:

1066

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

1067

client.attributes["Enabled"] = False

1068

1069

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

1070

1071

for clientpath in self.clients:

1072

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

1073

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

1074

1075

class TestDisableCmd(TestCmd):

1076

def test_disable(self):

1077

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

1078

for clientpath in self.clients:

1079

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

1080

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

1081

1082

class Unique(object):

1083

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

1084

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

1085

1086

class TestPropertyCmd(TestCmd):

1087

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

1088

def runTest(self):

1089

if not hasattr(self, "command"):

1090

return

1091

values_to_get = getattr(self, "values_to_get",

1092

self.values_to_set)

1093

for value_to_set, value_to_get in zip(self.values_to_set,

1094

values_to_get):

1095

for clientpath in self.clients:

1096

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

1097

old_value = client.attributes[self.propname]

1098

self.assertNotIsInstance(old_value, Unique)

1099

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

1100

self.run_command(value_to_set, self.clients)

1101

for clientpath in self.clients:

1102

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

1103

value = client.attributes[self.propname]

1104

self.assertNotIsInstance(value, Unique)

1105

self.assertEqual(value, value_to_get)

1106

def run_command(self, value, clients):

1107

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

1108

1109

class TestBumpTimeoutCmd(TestPropertyCmd):

1110

command = BumpTimeoutCmd

1111

propname = "LastCheckedOK"

1112

values_to_set = [""]

1113

1114

class TestStartCheckerCmd(TestPropertyCmd):

1115

command = StartCheckerCmd

1116

propname = "CheckerRunning"

1117

values_to_set = [dbus.Boolean(True)]

1118

1119

class TestStopCheckerCmd(TestPropertyCmd):

1120

command = StopCheckerCmd

1121

propname = "CheckerRunning"

1122

values_to_set = [dbus.Boolean(False)]

1123

1124

class TestApproveByDefaultCmd(TestPropertyCmd):

1125

command = ApproveByDefaultCmd

1126

propname = "ApprovedByDefault"

1127

values_to_set = [dbus.Boolean(True)]

1128

1129

class TestDenyByDefaultCmd(TestPropertyCmd):

1130

command = DenyByDefaultCmd

1131

propname = "ApprovedByDefault"

1132

values_to_set = [dbus.Boolean(False)]

1133

1134

class TestPropertyValueCmd(TestPropertyCmd):

1135

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

1136

def runTest(self):

1137

if type(self) is TestPropertyValueCmd:

1138

return

1139

return super(TestPropertyValueCmd, self).runTest()

1140

def run_command(self, value, clients):

1141

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

1142

1143

class TestSetCheckerCmd(TestPropertyValueCmd):

1144

command = SetCheckerCmd

1145

propname = "Checker"

1146

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

1147

1148

class TestSetHostCmd(TestPropertyValueCmd):

1149

command = SetHostCmd

1150

propname = "Host"

1151

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

1152

1153

class TestSetSecretCmd(TestPropertyValueCmd):

1154

command = SetSecretCmd

1155

propname = "Secret"

1156

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

1157

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

1158

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

1159

1160

class TestSetTimeoutCmd(TestPropertyValueCmd):

1161

command = SetTimeoutCmd

1162

propname = "Timeout"

1163

values_to_set = [datetime.timedelta(),

1164

datetime.timedelta(minutes=5),

1165

datetime.timedelta(seconds=1),

1166

datetime.timedelta(weeks=1),

1167

datetime.timedelta(weeks=52)]

1168

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

1169

1170

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1171

command = SetExtendedTimeoutCmd

1172

propname = "ExtendedTimeout"

1173

values_to_set = [datetime.timedelta(),

1174

datetime.timedelta(minutes=5),

1175

datetime.timedelta(seconds=1),

1176

datetime.timedelta(weeks=1),

1177

datetime.timedelta(weeks=52)]

1178

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

1179

1180

class TestSetIntervalCmd(TestPropertyValueCmd):

1181

command = SetIntervalCmd

1182

propname = "Interval"

1183

values_to_set = [datetime.timedelta(),

1184

datetime.timedelta(minutes=5),

1185

datetime.timedelta(seconds=1),

1186

datetime.timedelta(weeks=1),

1187

datetime.timedelta(weeks=52)]

1188

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

1189

1190

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1191

command = SetApprovalDelayCmd

1192

propname = "ApprovalDelay"

1193

values_to_set = [datetime.timedelta(),

1194

datetime.timedelta(minutes=5),

1195

datetime.timedelta(seconds=1),

1196

datetime.timedelta(weeks=1),

1197

datetime.timedelta(weeks=52)]

1198

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

1199

1200

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1201

command = SetApprovalDurationCmd

1202

propname = "ApprovalDuration"

1203

values_to_set = [datetime.timedelta(),

1204

datetime.timedelta(minutes=5),

1205

datetime.timedelta(seconds=1),

1206

datetime.timedelta(weeks=1),

1207

datetime.timedelta(weeks=52)]

1208

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

1209

1210

class Test_command_from_options(unittest.TestCase):

1211

def setUp(self):

1212

self.parser = argparse.ArgumentParser()

1213

add_command_line_options(self.parser)

1214

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

1215

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

1216

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

1217

options = self.parser.parse_args(args)

1218

check_option_syntax(self.parser, options)

1219

commands = commands_from_options(options)

1220

self.assertEqual(len(commands), 1)

1221

command = commands[0]

1222

self.assertIsInstance(command, command_cls)

1223

for key, value in cmd_attrs.items():

1224

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

1225

def test_print_table(self):

1226

self.assert_command_from_args([], PrintTableCmd,

1227

verbose=False)

1228

1229

def test_print_table_verbose(self):

1230

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

1231

verbose=True)

1232

1233

def test_print_table_verbose_short(self):

1234

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

1235

verbose=True)

1236

1237

def test_enable(self):

1238

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

1239

1240

def test_enable_short(self):

1241

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

1242

1243

def test_disable(self):

1244

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

1245

DisableCmd)

1246

1247

def test_disable_short(self):

1248

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

1249

1250

def test_bump_timeout(self):

1251

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

1252

BumpTimeoutCmd)

1253

1254

def test_bump_timeout_short(self):

1255

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

1256

1257

def test_start_checker(self):

1258

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

1259

StartCheckerCmd)

1260

1261

def test_stop_checker(self):

1262

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

1263

StopCheckerCmd)

1264

1265

def test_remove(self):

1266

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

1267

RemoveCmd)

1268

1269

def test_remove_short(self):

1270

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

1271

1272

def test_checker(self):

1273

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

1274

SetCheckerCmd, value_to_set=":")

1275

1276

def test_checker_empty(self):

1277

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

1278

SetCheckerCmd, value_to_set="")

1279

1280

def test_checker_short(self):

1281

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

1282

SetCheckerCmd, value_to_set=":")

1283

1284

def test_timeout(self):

1285

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

1286

SetTimeoutCmd,

1287

value_to_set=300000)

1288

1289

def test_timeout_short(self):

1290

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

1291

SetTimeoutCmd,

1292

value_to_set=300000)

1293

1294

def test_extended_timeout(self):

1295

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

1296

"foo"],

1297

SetExtendedTimeoutCmd,

1298

value_to_set=900000)

1299

1300

def test_interval(self):

1301

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

1302

SetIntervalCmd,

1303

value_to_set=120000)

1304

1305

def test_interval_short(self):

1306

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

1307

SetIntervalCmd,

1308

value_to_set=120000)

1309

1310

def test_approve_by_default(self):

1311

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

1312

ApproveByDefaultCmd)

1313

1314

def test_deny_by_default(self):

1315

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

1316

DenyByDefaultCmd)

1317

1318

def test_approval_delay(self):

1319

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

1320

"foo"], SetApprovalDelayCmd,

1321

value_to_set=30000)

1322

1323

def test_approval_duration(self):

1324

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

1325

"foo"], SetApprovalDurationCmd,

1326

value_to_set=1000)

1327

1328

def test_host(self):

1329

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

1330

"foo"], SetHostCmd,

1331

value_to_set="foo.example.org")

1332

1333

def test_host_short(self):

1334

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

1335

"foo"], SetHostCmd,

1336

value_to_set="foo.example.org")

1337

1338

def test_secret_devnull(self):

1339

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

1340

"foo"], SetSecretCmd,

1341

value_to_set=b"")

1342

1343

def test_secret_tempfile(self):

1344

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

1345

value = b"secret\0xyzzy\nbar"

1346

f.write(value)

1347

f.seek(0)

1348

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

1349

"foo"], SetSecretCmd,

1350

value_to_set=value)

1351

1352

def test_secret_devnull_short(self):

1353

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

1354

SetSecretCmd, value_to_set=b"")

1355

1356

def test_secret_tempfile_short(self):

1357

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

1358

value = b"secret\0xyzzy\nbar"

1359

f.write(value)

1360

f.seek(0)

1361

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

1362

SetSecretCmd,

1363

value_to_set=value)

1364

1365

def test_approve(self):

1366

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

1367

ApproveCmd)

1368

1369

def test_approve_short(self):

1370

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

1371

1372

def test_deny(self):

1373

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

1374

1375

def test_deny_short(self):

1376

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

1377

1378

def test_dump_json(self):

1379

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

1380

1381

def test_is_enabled(self):

1382

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

1383

IsEnabledCmd)

1384

1385

def test_is_enabled_short(self):

1386

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

1387

1388

def test_deny_before_remove(self):

1389

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

1390

check_option_syntax(self.parser, options)

1391

commands = commands_from_options(options)

1392

self.assertEqual(len(commands), 2)

1393

self.assertIsInstance(commands[0], DenyCmd)

1394

self.assertIsInstance(commands[1], RemoveCmd)

1395

1396

def test_deny_before_remove_reversed(self):

1397

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

1398

check_option_syntax(self.parser, options)

1399

commands = commands_from_options(options)

1400

self.assertEqual(len(commands), 2)

1401

self.assertIsInstance(commands[0], DenyCmd)

1402

self.assertIsInstance(commands[1], RemoveCmd)

1403

1404

1405

class Test_check_option_syntax(unittest.TestCase):

1406

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

1407

# check_option_syntax()

1408

actions = {

1409

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

1410

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

1411

# never used

1412

"enable": True,

1413

"disable": True,

1414

"bump_timeout": True,

1415

"start_checker": True,

1416

"stop_checker": True,

1417

"is_enabled": True,

1418

"remove": True,

1419

"checker": "x",

1420

"timeout": datetime.timedelta(),

1421

"extended_timeout": datetime.timedelta(),

1422

"interval": datetime.timedelta(),

1423

"approved_by_default": True,

1424

"approval_delay": datetime.timedelta(),

1425

"approval_duration": datetime.timedelta(),

1426

"host": "x",

1427

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

1428

"approve": True,

1429

"deny": True,

1430

}

1431

1432

def setUp(self):

1433

self.parser = argparse.ArgumentParser()

1434

add_command_line_options(self.parser)

1435

1436

@contextlib.contextmanager

1437

def assertParseError(self):

1438

with self.assertRaises(SystemExit) as e:

1439

with self.temporarily_suppress_stderr():

1440

yield

1441

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

1442

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

1443

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

1444

1445

@staticmethod

1446

@contextlib.contextmanager

1447

def temporarily_suppress_stderr():

1448

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

1449

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

1450

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

1451

os.close(null)

1452

try:

1453

yield

1454

finally:

1455

# restore stderr

1456

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

1457

os.close(stderrcopy)

1458

1459

def check_option_syntax(self, options):

1460

check_option_syntax(self.parser, options)

1461

1462

def test_actions_requires_client_or_all(self):

1463

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

1464

options = self.parser.parse_args()

1465

setattr(options, action, value)

1466

with self.assertParseError():

1467

self.check_option_syntax(options)

1468

1469

def test_actions_conflicts_with_verbose(self):

1470

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

1471

options = self.parser.parse_args()

1472

setattr(options, action, value)

1473

options.verbose = True

1474

with self.assertParseError():

1475

self.check_option_syntax(options)

1476

1477

def test_dump_json_conflicts_with_verbose(self):

1478

options = self.parser.parse_args()

1479

options.dump_json = True

1480

options.verbose = True

1481

with self.assertParseError():

1482

self.check_option_syntax(options)

1483

1484

def test_dump_json_conflicts_with_action(self):

1485

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

1486

options = self.parser.parse_args()

1487

setattr(options, action, value)

1488

options.dump_json = True

1489

with self.assertParseError():

1490

self.check_option_syntax(options)

1491

1492

def test_all_can_not_be_alone(self):

1493

options = self.parser.parse_args()

1494

options.all = True

1495

with self.assertParseError():

1496

self.check_option_syntax(options)

1497

1498

def test_all_is_ok_with_any_action(self):

1499

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

1500

options = self.parser.parse_args()

1501

setattr(options, action, value)

1502

options.all = True

1503

self.check_option_syntax(options)

1504

1505

def test_is_enabled_fails_without_client(self):

1506

options = self.parser.parse_args()

1507

options.is_enabled = True

1508

with self.assertParseError():

1509

self.check_option_syntax(options)

1510

1511

def test_is_enabled_works_with_one_client(self):

1512

options = self.parser.parse_args()

1513

options.is_enabled = True

1514

options.client = ["foo"]

1515

self.check_option_syntax(options)

1516

1517

def test_is_enabled_fails_with_two_clients(self):

1518

options = self.parser.parse_args()

1519

options.is_enabled = True

1520

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

1521

with self.assertParseError():

1522

self.check_option_syntax(options)

1523

1524

def test_remove_can_only_be_combined_with_action_deny(self):

1525

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

1526

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

1527

continue

1528

options = self.parser.parse_args()

1529

setattr(options, action, value)

1530

options.all = True

1531

options.remove = True

1532

with self.assertParseError():

1533

self.check_option_syntax(options)

1534

1535

1536

1537

def should_only_run_tests():

1538

parser = argparse.ArgumentParser(add_help=False)

1539

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

1540

args, unknown_args = parser.parse_known_args()

1541

run_tests = args.check

1542

if run_tests:

1543

# Remove --check argument from sys.argv

1544

sys.argv[1:] = unknown_args

1545

return run_tests

1546

1547

# Add all tests from doctest strings

1548

def load_tests(loader, tests, none):

1549

import doctest

1550

tests.addTests(doctest.DocTestSuite())

1551

return tests

1552

1553

if __name__ == "__main__":

1554

if should_only_run_tests():

1555

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

1556

unittest.main()

1557

else:

1558

main()