/mandos/release : revision 237.7.619

1

#!/usr/bin/python

2

# -*- mode: python; coding: utf-8 -*-

3

#

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 program is free software: you can redistribute it and/or modify

10

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

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

11

13

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

12

14

# (at your option) any later version.

13

15

#

14

# This program is distributed in the hope that it will be useful,

15

# but WITHOUT ANY WARRANTY; without even the implied warranty of

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

16

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17

19

# GNU General Public License for more details.

18

#

20

#

19

21

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

20

# along with this program. If not, see

21

# <http://www.gnu.org/licenses/>.

22

#

22

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

23

#

23

24

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

24

#

25

#

25

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

29

35

import sys

30

import dbus

31

36

import argparse

32

37

import locale

33

38

import datetime

34

39

import re

35

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

36

64

37

65

locale.setlocale(locale.LC_ALL, "")

38

66

39

tablewords = {

40

"Name": "Name",

41

"Enabled": "Enabled",

42

"Timeout": "Timeout",

43

"LastCheckedOK": "Last Successful Check",

44

"LastApprovalRequest": "Last Approval Request",

45

"Created": "Created",

46

"Interval": "Interval",

47

"Host": "Host",

48

"Fingerprint": "Fingerprint",

49

"CheckerRunning": "Check Is Running",

50

"LastEnabled": "Last Enabled",

51

"ApprovalPending": "Approval Is Pending",

52

"ApprovedByDefault": "Approved By Default",

53

"ApprovalDelay": "Approval Delay",

54

"ApprovalDuration": "Approval Duration",

55

"Checker": "Checker",

56

"ExtendedTimeout" : "Extended Timeout"

57

}

58

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

59

domain = "se.recompile"

60

busname = domain + ".Mandos"

61

server_path = "/"

62

server_interface = domain + ".Mandos"

63

client_interface = domain + ".Mandos.Client"

64

version = "1.5.3"

65

66

def timedelta_to_milliseconds(td):

67

"""Convert a datetime.timedelta object to milliseconds"""

68

return ((td.days * 24 * 60 * 60 * 1000)

69

+ (td.seconds * 1000)

70

+ (td.microseconds // 1000))

71

72

def milliseconds_to_string(ms):

73

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

74

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

75

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

76

hours = td.seconds // 3600,

77

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

78

seconds = td.seconds % 60,

79

))

80

81

def string_to_delta(interval):

82

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

83

84

>>> string_to_delta("7d")

85

datetime.timedelta(7)

86

>>> string_to_delta("60s")

87

datetime.timedelta(0, 60)

88

>>> string_to_delta("60m")

89

datetime.timedelta(0, 3600)

90

>>> string_to_delta("24h")

91

datetime.timedelta(1)

92

>>> string_to_delta("1w")

93

datetime.timedelta(7)

94

>>> string_to_delta("5m 30s")

95

datetime.timedelta(0, 330)

96

"""

97

value = datetime.timedelta(0)

98

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

99

100

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

101

if suffix == "d":

102

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

103

elif suffix == "s":

104

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

105

elif suffix == "m":

106

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

107

elif suffix == "h":

108

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

109

elif suffix == "w":

110

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

111

elif suffix == "":

112

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

113

return value

114

115

def print_clients(clients, keywords):

116

def valuetostring(value, keyword):

117

if type(value) is dbus.Boolean:

118

return "Yes" if value else "No"

119

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

120

"ApprovalDuration", "ExtendedTimeout"):

121

return milliseconds_to_string(value)

122

return unicode(value)

123

124

# Create format string to print table rows

125

format_string = " ".join("{{{key}:{width}}}".format(

126

width = max(len(tablewords[key]),

127

max(len(valuetostring(client[key],

128

key))

129

for client in

130

clients)),

131

key = key) for key in keywords)

132

# Print header line

133

print(format_string.format(**tablewords))

134

for client in clients:

135

print(format_string.format(**dict((key,

136

valuetostring(client[key],

137

key))

138

for key in keywords)))

139

140

def has_actions(options):

141

return any((options.enable,

142

options.disable,

143

options.bump_timeout,

144

options.start_checker,

145

options.stop_checker,

146

options.is_enabled,

147

options.remove,

148

options.checker is not None,

149

options.timeout is not None,

150

options.extended_timeout is not None,

151

options.interval is not None,

152

options.approved_by_default is not None,

153

options.approval_delay is not None,

154

options.approval_duration is not None,

155

options.host is not None,

156

options.secret is not None,

157

options.approve,

158

options.deny))

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

159

81

160

82

def main():

161

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

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

126

sys.exit(1)

127

128

# Run all commands on clients

129

commands = commands_from_options(options)

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

162

135

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

163

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

136

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

164

137

help="show version number and exit")

165

138

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

166

139

help="Select all clients")

167

140

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

168

141

help="Print all fields")

169

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

170

help="Enable client")

171

parser.add_argument("-d", "--disable", action="store_true",

172

help="disable client")

142

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

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

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

146

help="Enable client")

147

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

148

action="store_true",

149

help="disable client")

173

150

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

174

151

help="Bump timeout for client")

175

parser.add_argument("--start-checker", action="store_true",

176

help="Start checker for client")

177

parser.add_argument("--stop-checker", action="store_true",

178

help="Stop checker for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

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

154

action="store_true",

155

help="Start checker for client")

156

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

157

action="store_true",

158

help="Stop checker for client")

179

159

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

180

160

help="Check if client is enabled")

181

161

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

182

162

help="Remove client")

183

163

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

184

164

help="Set checker command for client")

185

parser.add_argument("-t", "--timeout",

165

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

186

166

help="Set timeout for client")

187

parser.add_argument("--extended-timeout",

167

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

188

168

help="Set extended timeout for client")

189

parser.add_argument("-i", "--interval",

169

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

190

170

help="Set checker interval for client")

191

parser.add_argument("--approve-by-default", action="store_true",

192

default=None, dest="approved_by_default",

193

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

194

parser.add_argument("--deny-by-default", action="store_false",

195

dest="approved_by_default",

196

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

197

parser.add_argument("--approval-delay",

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

175

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

176

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

179

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

180

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

198

181

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

199

parser.add_argument("--approval-duration",

182

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

200

183

help="Set duration of one client approval")

201

184

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

202

parser.add_argument("-s", "--secret", type=file,

185

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

186

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

203

187

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

204

parser.add_argument("-A", "--approve", action="store_true",

205

help="Approve any current client request")

206

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

207

help="Deny any current client request")

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

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

191

help="Approve any current client request")

192

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

193

help="Deny any current client request")

194

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

195

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

196

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

197

help="Run self-test")

208

198

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

209

options = parser.parse_args()

210

199

200

201

def string_to_delta(interval):

202

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

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

213

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

214

215

>>> rfc3339_duration_to_delta("P7D")

216

datetime.timedelta(7)

217

>>> rfc3339_duration_to_delta("PT60S")

218

datetime.timedelta(0, 60)

219

>>> rfc3339_duration_to_delta("PT60M")

220

datetime.timedelta(0, 3600)

221

>>> rfc3339_duration_to_delta("P60M")

222

datetime.timedelta(1680)

223

>>> rfc3339_duration_to_delta("PT24H")

224

datetime.timedelta(1)

225

>>> rfc3339_duration_to_delta("P1W")

226

datetime.timedelta(7)

227

>>> rfc3339_duration_to_delta("PT5M30S")

228

datetime.timedelta(0, 330)

229

>>> rfc3339_duration_to_delta("P1DT3M20S")

230

datetime.timedelta(1, 200)

231

>>> # Can not be empty:

232

>>> rfc3339_duration_to_delta("")

233

Traceback (most recent call last):

234

...

235

ValueError: Invalid RFC 3339 duration: u''

236

>>> # Must start with "P":

237

>>> rfc3339_duration_to_delta("1D")

238

Traceback (most recent call last):

239

...

240

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

241

>>> # Must use correct order

242

>>> rfc3339_duration_to_delta("PT1S2M")

243

Traceback (most recent call last):

244

...

245

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

246

>>> # Time needs time marker

247

>>> rfc3339_duration_to_delta("P1H2S")

248

Traceback (most recent call last):

249

...

250

ValueError: Invalid RFC 3339 duration: u'P1H2S'

251

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

252

>>> rfc3339_duration_to_delta("P1D2W")

253

Traceback (most recent call last):

254

...

255

ValueError: Invalid RFC 3339 duration: u'P1D2W'

256

>>> rfc3339_duration_to_delta("P2W2H")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: u'P2W2H'

260

"""

261

262

# Parsing an RFC 3339 duration with regular expressions is not

263

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

264

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

265

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

266

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

267

# avoid excessive use of external libraries.

268

269

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

270

Token = collections.namedtuple("Token", (

271

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

272

# a "group" containing digits

273

"value", # datetime.timedelta or None

274

"followers")) # Tokens valid after this token

275

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

276

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

277

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

278

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

279

datetime.timedelta(seconds=1),

280

frozenset((token_end, )))

281

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

282

datetime.timedelta(minutes=1),

283

frozenset((token_second, token_end)))

284

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

285

datetime.timedelta(hours=1),

286

frozenset((token_minute, token_end)))

287

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

288

None,

289

frozenset((token_hour, token_minute,

290

token_second)))

291

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

292

datetime.timedelta(days=1),

293

frozenset((token_time, token_end)))

294

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

295

datetime.timedelta(weeks=4),

296

frozenset((token_day, token_end)))

297

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

298

datetime.timedelta(weeks=52),

299

frozenset((token_month, token_end)))

300

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

301

datetime.timedelta(weeks=1),

302

frozenset((token_end, )))

303

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

304

frozenset((token_year, token_month,

305

token_day, token_time,

306

token_week)))

307

# Define starting values:

308

# Value so far

309

value = datetime.timedelta()

310

found_token = None

311

# Following valid tokens

312

followers = frozenset((token_duration, ))

313

# String left to parse

314

s = duration

315

# Loop until end token is found

316

while found_token is not token_end:

317

# Search for any currently valid tokens

318

for token in followers:

319

match = token.regexp.match(s)

320

if match is not None:

321

# Token found

322

if token.value is not None:

323

# Value found, parse digits

324

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

325

# Add to value so far

326

value += factor * token.value

327

# Strip token from string

328

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

329

# Go to found token

330

found_token = token

331

# Set valid next tokens

332

followers = found_token.followers

333

break

334

else:

335

# No currently valid tokens were found

336

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

337

.format(duration))

338

# End token found

339

return value

340

341

342

def parse_pre_1_6_1_interval(interval):

343

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

344

and return a datetime.timedelta

345

346

>>> parse_pre_1_6_1_interval('7d')

347

datetime.timedelta(7)

348

>>> parse_pre_1_6_1_interval('60s')

349

datetime.timedelta(0, 60)

350

>>> parse_pre_1_6_1_interval('60m')

351

datetime.timedelta(0, 3600)

352

>>> parse_pre_1_6_1_interval('24h')

353

datetime.timedelta(1)

354

>>> parse_pre_1_6_1_interval('1w')

355

datetime.timedelta(7)

356

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

357

datetime.timedelta(0, 330)

358

>>> parse_pre_1_6_1_interval('')

359

datetime.timedelta(0)

360

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

361

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

362

datetime.timedelta(2, 480, 18000)

363

364

"""

365

366

value = datetime.timedelta(0)

367

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

368

369

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

370

if suffix == "d":

371

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

372

elif suffix == "s":

373

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

374

elif suffix == "m":

375

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

376

elif suffix == "h":

377

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

378

elif suffix == "w":

379

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

380

elif suffix == "":

381

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

382

return value

383

384

385

def check_option_syntax(parser, options):

386

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

387

argparse"""

388

389

def has_actions(options):

390

return any((options.enable,

391

options.disable,

392

options.bump_timeout,

393

options.start_checker,

394

options.stop_checker,

395

options.is_enabled,

396

options.remove,

397

options.checker is not None,

398

options.timeout is not None,

399

options.extended_timeout is not None,

400

options.interval is not None,

401

options.approved_by_default is not None,

402

options.approval_delay is not None,

403

options.approval_duration is not None,

404

options.host is not None,

405

options.secret is not None,

406

options.approve,

407

options.deny))

408

211

409

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

212

410

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

213

411

if options.verbose and has_actions(options):

214

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

215

" --all.")

412

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

415

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

216

416

if options.all and not has_actions(options):

217

417

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

218

219

try:

220

bus = dbus.SystemBus()

221

mandos_dbus_objc = bus.get_object(busname, server_path)

222

except dbus.exceptions.DBusException:

223

print("Could not connect to Mandos server",

224

file=sys.stderr)

225

sys.exit(1)

226

227

mandos_serv = dbus.Interface(mandos_dbus_objc,

228

dbus_interface = server_interface)

229

230

#block stderr since dbus library prints to stderr

231

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

232

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

233

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

234

os.close(null)

235

try:

418

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

419

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

420

if options.remove:

421

options.remove = False

422

if has_actions(options) and not options.deny:

423

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

424

options.remove = True

425

426

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

"Could not connect to Mandos server: %s"):

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

log.critical(*(args + (e,)), **kwargs)

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

"Failed to access Mandos server through D-Bus:\n%s"):

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

def __exit__(self, exc_type, exc_val, exc_tb):

472

self.logger.removeFilter(self.nullfilter)

473

474

475

def commands_from_options(options):

476

477

commands = []

478

479

if options.is_enabled:

480

commands.append(IsEnabledCmd())

481

482

if options.approve:

483

commands.append(ApproveCmd())

484

485

if options.deny:

486

commands.append(DenyCmd())

487

488

if options.remove:

489

commands.append(RemoveCmd())

490

491

if options.dump_json:

492

commands.append(DumpJSONCmd())

493

494

if options.enable:

495

commands.append(EnableCmd())

496

497

if options.disable:

498

commands.append(DisableCmd())

499

500

if options.bump_timeout:

501

commands.append(BumpTimeoutCmd())

502

503

if options.start_checker:

504

commands.append(StartCheckerCmd())

505

506

if options.stop_checker:

507

commands.append(StopCheckerCmd())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(ApproveByDefaultCmd())

512

else:

513

commands.append(DenyByDefaultCmd())

514

515

if options.checker is not None:

516

commands.append(SetCheckerCmd(options.checker))

517

518

if options.host is not None:

519

commands.append(SetHostCmd(options.host))

520

521

if options.secret is not None:

522

commands.append(SetSecretCmd(options.secret))

523

524

if options.timeout is not None:

525

commands.append(SetTimeoutCmd(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

SetExtendedTimeoutCmd(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(SetIntervalCmd(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(SetApprovalDelayCmd(options.approval_delay))

536

537

if options.approval_duration is not None:

538

commands.append(

539

SetApprovalDurationCmd(options.approval_duration))

540

541

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

542

# optionally verbosely

543

if not commands:

544

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

545

546

return commands

547

548

549

class Command(object):

550

"""Abstract class for commands"""

551

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

552

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

553

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

554

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

555

self.mandos = mandos

556

for clientpath, properties in clients.items():

557

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

558

dbus_busname, str(clientpath))

559

client = bus.get_object(dbus_busname, clientpath)

560

self.run_on_one_client(client, properties)

561

562

563

class IsEnabledCmd(Command):

564

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

565

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

566

if self.is_enabled(client, properties):

567

sys.exit(0)

568

sys.exit(1)

569

def is_enabled(self, client, properties):

570

return properties["Enabled"]

571

572

573

class ApproveCmd(Command):

574

def run_on_one_client(self, client, properties):

575

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

576

client.__dbus_object_path__, client_dbus_interface)

577

client.Approve(dbus.Boolean(True),

578

dbus_interface=client_dbus_interface)

579

580

581

class DenyCmd(Command):

582

def run_on_one_client(self, client, properties):

583

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

584

client.__dbus_object_path__, client_dbus_interface)

585

client.Approve(dbus.Boolean(False),

586

dbus_interface=client_dbus_interface)

587

588

589

class RemoveCmd(Command):

590

def run_on_one_client(self, client, properties):

591

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

592

server_dbus_path, server_dbus_interface,

593

str(client.__dbus_object_path__))

594

self.mandos.RemoveClient(client.__dbus_object_path__)

595

596

597

class OutputCmd(Command):

598

"""Abstract class for commands outputting client details"""

599

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

600

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

601

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

602

"ApprovalPending", "ApprovedByDefault",

603

"LastApprovalRequest", "ApprovalDelay",

604

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

605

"Expires", "LastCheckerStatus")

606

607

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

608

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

609

610

def output(self, clients):

611

raise NotImplementedError()

612

613

614

class DumpJSONCmd(OutputCmd):

615

def output(self, clients):

616

data = {client["Name"]:

617

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

618

for key in self.all_keywords}

619

for client in clients}

620

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

621

622

@staticmethod

623

def dbus_boolean_to_bool(value):

624

if isinstance(value, dbus.Boolean):

625

value = bool(value)

626

return value

627

628

629

class PrintTableCmd(OutputCmd):

630

def __init__(self, verbose=False):

631

self.verbose = verbose

632

633

def output(self, clients):

634

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

635

"LastCheckedOK")

636

keywords = default_keywords

637

if self.verbose:

638

keywords = self.all_keywords

639

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

640

641

class TableOfClients(object):

642

tableheaders = {

643

"Name": "Name",

644

"Enabled": "Enabled",

645

"Timeout": "Timeout",

646

"LastCheckedOK": "Last Successful Check",

647

"LastApprovalRequest": "Last Approval Request",

648

"Created": "Created",

649

"Interval": "Interval",

650

"Host": "Host",

651

"Fingerprint": "Fingerprint",

652

"KeyID": "Key ID",

653

"CheckerRunning": "Check Is Running",

654

"LastEnabled": "Last Enabled",

655

"ApprovalPending": "Approval Is Pending",

656

"ApprovedByDefault": "Approved By Default",

657

"ApprovalDelay": "Approval Delay",

658

"ApprovalDuration": "Approval Duration",

659

"Checker": "Checker",

660

"ExtendedTimeout": "Extended Timeout",

661

"Expires": "Expires",

662

"LastCheckerStatus": "Last Checker Status",

663

}

664

665

def __init__(self, clients, keywords):

666

self.clients = clients

667

self.keywords = keywords

668

669

def __str__(self):

670

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

671

672

if sys.version_info.major == 2:

673

__unicode__ = __str__

674

def __str__(self):

675

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

676

677

def rows(self):

678

format_string = self.row_formatting_string()

679

rows = [self.header_line(format_string)]

680

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

681

for client in self.clients)

682

return rows

683

684

def row_formatting_string(self):

685

"Format string used to format table rows"

686

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

687

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

688

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

689

for client in self.clients)),

690

key=key)

691

for key in self.keywords)

692

693

def string_from_client(self, client, key):

694

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

695

696

@classmethod

697

def valuetostring(cls, value, keyword):

698

if isinstance(value, dbus.Boolean):

699

return "Yes" if value else "No"

700

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

701

"ApprovalDuration", "ExtendedTimeout"):

702

return cls.milliseconds_to_string(value)

703

return str(value)

704

705

def header_line(self, format_string):

706

return format_string.format(**self.tableheaders)

707

708

def client_line(self, client, format_string):

709

return format_string.format(

710

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

711

for key in self.keywords})

712

713

@staticmethod

714

def milliseconds_to_string(ms):

715

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

716

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

717

.format(days="{}T".format(td.days)

718

if td.days else "",

719

hours=td.seconds // 3600,

720

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

721

seconds=td.seconds % 60))

722

723

724

class PropertyCmd(Command):

725

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

726

727

def run_on_one_client(self, client, properties):

728

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

729

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

730

client.__dbus_object_path__,

731

dbus.PROPERTIES_IFACE, client_dbus_interface,

732

self.propname, self.value_to_set

733

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

734

else bool(self.value_to_set))

735

client.Set(client_dbus_interface, self.propname,

736

self.value_to_set,

737

dbus_interface=dbus.PROPERTIES_IFACE)

738

739

@property

740

def propname(self):

741

raise NotImplementedError()

742

743

744

class EnableCmd(PropertyCmd):

745

propname = "Enabled"

746

value_to_set = dbus.Boolean(True)

747

748

749

class DisableCmd(PropertyCmd):

750

propname = "Enabled"

751

value_to_set = dbus.Boolean(False)

752

753

754

class BumpTimeoutCmd(PropertyCmd):

755

propname = "LastCheckedOK"

756

value_to_set = ""

757

758

759

class StartCheckerCmd(PropertyCmd):

760

propname = "CheckerRunning"

761

value_to_set = dbus.Boolean(True)

762

763

764

class StopCheckerCmd(PropertyCmd):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(False)

767

768

769

class ApproveByDefaultCmd(PropertyCmd):

770

propname = "ApprovedByDefault"

771

value_to_set = dbus.Boolean(True)

772

773

774

class DenyByDefaultCmd(PropertyCmd):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(False)

777

778

779

class PropertyValueCmd(PropertyCmd):

780

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

781

def __init__(self, value):

782

self.value_to_set = value

783

784

785

class SetCheckerCmd(PropertyValueCmd):

786

propname = "Checker"

787

788

789

class SetHostCmd(PropertyValueCmd):

790

propname = "Host"

791

792

793

class SetSecretCmd(PropertyValueCmd):

794

propname = "Secret"

795

796

@property

797

def value_to_set(self):

798

return self._vts

799

800

@value_to_set.setter

801

def value_to_set(self, value):

802

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

803

self._vts = value.read()

804

value.close()

805

806

807

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

808

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

809

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

810

811

@property

812

def value_to_set(self):

813

return self._vts

814

815

@value_to_set.setter

816

def value_to_set(self, value):

817

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

818

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

819

820

821

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

822

propname = "Timeout"

823

824

825

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

826

propname = "ExtendedTimeout"

827

828

829

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

830

propname = "Interval"

831

832

833

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

834

propname = "ApprovalDelay"

835

836

837

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

838

propname = "ApprovalDuration"

839

840

841

842

class Test_string_to_delta(unittest.TestCase):

843

def test_handles_basic_rfc3339(self):

844

self.assertEqual(string_to_delta("PT0S"),

845

datetime.timedelta())

846

self.assertEqual(string_to_delta("P0D"),

847

datetime.timedelta())

848

self.assertEqual(string_to_delta("PT1S"),

849

datetime.timedelta(0, 1))

850

self.assertEqual(string_to_delta("PT2H"),

851

datetime.timedelta(0, 7200))

852

853

def test_falls_back_to_pre_1_6_1_with_warning(self):

854

# assertLogs only exists in Python 3.4

855

if hasattr(self, "assertLogs"):

856

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

857

value = string_to_delta("2h")

858

else:

859

class WarningFilter(logging.Filter):

860

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

861

def filter(self, record):

862

is_warning = record.levelno >= logging.WARNING

863

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

864

False)

865

return not is_warning

866

warning_filter = WarningFilter()

867

log.addFilter(warning_filter)

868

try:

869

value = string_to_delta("2h")

870

finally:

871

log.removeFilter(warning_filter)

872

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

873

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

874

875

876

class Test_check_option_syntax(unittest.TestCase):

877

def setUp(self):

878

self.parser = argparse.ArgumentParser()

879

add_command_line_options(self.parser)

880

881

def test_actions_requires_client_or_all(self):

882

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

883

options = self.parser.parse_args()

884

setattr(options, action, value)

885

with self.assertParseError():

886

self.check_option_syntax(options)

887

888

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

889

# check_option_syntax()

890

actions = {

891

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

892

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

893

# never used

894

"enable": True,

895

"disable": True,

896

"bump_timeout": True,

897

"start_checker": True,

898

"stop_checker": True,

899

"is_enabled": True,

900

"remove": True,

901

"checker": "x",

902

"timeout": datetime.timedelta(),

903

"extended_timeout": datetime.timedelta(),

904

"interval": datetime.timedelta(),

905

"approved_by_default": True,

906

"approval_delay": datetime.timedelta(),

907

"approval_duration": datetime.timedelta(),

908

"host": "x",

909

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

910

"approve": True,

911

"deny": True,

912

}

913

914

@contextlib.contextmanager

915

def assertParseError(self):

916

with self.assertRaises(SystemExit) as e:

917

with self.temporarily_suppress_stderr():

918

yield

919

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

920

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

921

# /argparse.html#exiting-methods

922

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

923

924

@staticmethod

925

@contextlib.contextmanager

926

def temporarily_suppress_stderr():

927

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

928

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

929

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

930

os.close(null)

236

931

try:

237

mandos_clients = mandos_serv.GetAllClientsWithProperties()

932

yield

238

933

finally:

239

#restore stderr

934

# restore stderr

240

935

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

241

936

os.close(stderrcopy)

242

except dbus.exceptions.DBusException:

243

print("Access denied: Accessing mandos server through dbus.",

244

file=sys.stderr)

245

sys.exit(1)

246

247

# Compile dict of (clients: properties) to process

248

clients={}

249

250

if options.all or not options.client:

251

clients = dict((bus.get_object(busname, path), properties)

252

for path, properties in

253

mandos_clients.iteritems())

254

else:

255

for name in options.client:

256

for path, client in mandos_clients.iteritems():

257

if client["Name"] == name:

258

client_objc = bus.get_object(busname, path)

259

clients[client_objc] = client

260

break

261

else:

262

print("Client not found on server: {0!r}"

263

.format(name), file=sys.stderr)

264

sys.exit(1)

265

266

if not has_actions(options) and clients:

267

if options.verbose:

268

keywords = ("Name", "Enabled", "Timeout",

269

"LastCheckedOK", "Created", "Interval",

270

"Host", "Fingerprint", "CheckerRunning",

271

"LastEnabled", "ApprovalPending",

272

"ApprovedByDefault",

273

"LastApprovalRequest", "ApprovalDelay",

274

"ApprovalDuration", "Checker",

275

"ExtendedTimeout")

276

else:

277

keywords = defaultkeywords

278

279

print_clients(clients.values(), keywords)

280

else:

281

# Process each client in the list by all selected options

282

for client in clients:

283

def set_client_prop(prop, value):

284

"""Set a Client D-Bus property"""

285

client.Set(client_interface, prop, value,

286

dbus_interface=dbus.PROPERTIES_IFACE)

287

def set_client_prop_ms(prop, value):

288

"""Set a Client D-Bus property, converted

289

from a string to milliseconds."""

290

set_client_prop(prop,

291

timedelta_to_milliseconds

292

(string_to_delta(value)))

293

if options.remove:

294

mandos_serv.RemoveClient(client.__dbus_object_path__)

295

if options.enable:

296

set_client_prop("Enabled", dbus.Boolean(True))

297

if options.disable:

298

set_client_prop("Enabled", dbus.Boolean(False))

299

if options.bump_timeout:

300

set_client_prop("LastCheckedOK", "")

301

if options.start_checker:

302

set_client_prop("CheckerRunning", dbus.Boolean(True))

303

if options.stop_checker:

304

set_client_prop("CheckerRunning", dbus.Boolean(False))

305

if options.is_enabled:

306

sys.exit(0 if client.Get(client_interface,

307

"Enabled",

308

dbus_interface=

309

dbus.PROPERTIES_IFACE)

310

else 1)

311

if options.checker is not None:

312

set_client_prop("Checker", options.checker)

313

if options.host is not None:

314

set_client_prop("Host", options.host)

315

if options.interval is not None:

316

set_client_prop_ms("Interval", options.interval)

317

if options.approval_delay is not None:

318

set_client_prop_ms("ApprovalDelay",

319

options.approval_delay)

320

if options.approval_duration is not None:

321

set_client_prop_ms("ApprovalDuration",

322

options.approval_duration)

323

if options.timeout is not None:

324

set_client_prop_ms("Timeout", options.timeout)

325

if options.extended_timeout is not None:

326

set_client_prop_ms("ExtendedTimeout",

327

options.extended_timeout)

328

if options.secret is not None:

329

set_client_prop("Secret",

330

dbus.ByteArray(options.secret.read()))

331

if options.approved_by_default is not None:

332

set_client_prop("ApprovedByDefault",

333

dbus.Boolean(options

334

.approved_by_default))

335

if options.approve:

336

client.Approve(dbus.Boolean(True),

337

dbus_interface=client_interface)

338

elif options.deny:

339

client.Approve(dbus.Boolean(False),

340

dbus_interface=client_interface)

937

938

def check_option_syntax(self, options):

939

check_option_syntax(self.parser, options)

940

941

def test_actions_conflicts_with_verbose(self):

942

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

943

options = self.parser.parse_args()

944

setattr(options, action, value)

945

options.verbose = True

946

with self.assertParseError():

947

self.check_option_syntax(options)

948

949

def test_dump_json_conflicts_with_verbose(self):

950

options = self.parser.parse_args()

951

options.dump_json = True

952

options.verbose = True

953

with self.assertParseError():

954

self.check_option_syntax(options)

955

956

def test_dump_json_conflicts_with_action(self):

957

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

958

options = self.parser.parse_args()

959

setattr(options, action, value)

960

options.dump_json = True

961

with self.assertParseError():

962

self.check_option_syntax(options)

963

964

def test_all_can_not_be_alone(self):

965

options = self.parser.parse_args()

966

options.all = True

967

with self.assertParseError():

968

self.check_option_syntax(options)

969

970

def test_all_is_ok_with_any_action(self):

971

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

972

options = self.parser.parse_args()

973

setattr(options, action, value)

974

options.all = True

975

self.check_option_syntax(options)

976

977

def test_is_enabled_fails_without_client(self):

978

options = self.parser.parse_args()

979

options.is_enabled = True

980

with self.assertParseError():

981

self.check_option_syntax(options)

982

983

def test_is_enabled_works_with_one_client(self):

984

options = self.parser.parse_args()

985

options.is_enabled = True

986

options.client = ["foo"]

987

self.check_option_syntax(options)

988

989

def test_is_enabled_fails_with_two_clients(self):

990

options = self.parser.parse_args()

991

options.is_enabled = True

992

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

993

with self.assertParseError():

994

self.check_option_syntax(options)

995

996

def test_remove_can_only_be_combined_with_action_deny(self):

997

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

998

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

999

continue

1000

options = self.parser.parse_args()

1001

setattr(options, action, value)

1002

options.all = True

1003

options.remove = True

1004

with self.assertParseError():

1005

self.check_option_syntax(options)

1006

1007

1008

class Test_get_mandos_dbus_object(unittest.TestCase):

1009

def test_calls_and_returns_get_object_on_bus(self):

1010

class MockBus(object):

1011

called = False

1012

def get_object(mockbus_self, busname, dbus_path):

1013

# Note that "self" is still the testcase instance,

1014

# this MockBus instance is in "mockbus_self".

1015

self.assertEqual(busname, dbus_busname)

1016

self.assertEqual(dbus_path, server_dbus_path)

1017

mockbus_self.called = True

1018

return mockbus_self

1019

1020

mockbus = get_mandos_dbus_object(bus=MockBus())

1021

self.assertIsInstance(mockbus, MockBus)

1022

self.assertTrue(mockbus.called)

1023

1024

def test_logs_and_exits_on_dbus_error(self):

1025

class MockBusFailing(object):

1026

def get_object(self, busname, dbus_path):

1027

raise dbus.exceptions.DBusException("Test")

1028

1029

# assertLogs only exists in Python 3.4

1030

if hasattr(self, "assertLogs"):

1031

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

1032

with self.assertRaises(SystemExit) as e:

1033

bus = get_mandos_dbus_object(bus=MockBus())

1034

else:

1035

critical_filter = self.CriticalFilter()

1036

log.addFilter(critical_filter)

1037

try:

1038

with self.assertRaises(SystemExit) as e:

1039

get_mandos_dbus_object(bus=MockBusFailing())

1040

finally:

1041

log.removeFilter(critical_filter)

1042

self.assertTrue(critical_filter.found)

1043

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

1044

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

1045

else:

1046

self.assertIsNotNone(e.exception.code)

1047

1048

class CriticalFilter(logging.Filter):

1049

"""Don't show, but register, critical messages"""

1050

found = False

1051

def filter(self, record):

1052

is_critical = record.levelno >= logging.CRITICAL

1053

self.found = is_critical or self.found

1054

return not is_critical

1055

1056

1057

class Test_get_managed_objects(unittest.TestCase):

1058

def test_calls_and_returns_GetManagedObjects(self):

1059

managed_objects = {"/clients/foo": { "Name": "foo"}}

1060

class MockObjectManager(object):

1061

@staticmethod

1062

def GetManagedObjects():

1063

return managed_objects

1064

retval = get_managed_objects(MockObjectManager())

1065

self.assertDictEqual(managed_objects, retval)

1066

1067

def test_logs_and_exits_on_dbus_error(self):

1068

class MockObjectManagerFailing(object):

1069

@staticmethod

1070

def GetManagedObjects():

1071

raise dbus.exceptions.DBusException("Test")

1072

1073

if hasattr(self, "assertLogs"):

1074

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

1075

with self.assertRaises(SystemExit):

1076

get_managed_objects(MockObjectManagerFailing())

1077

else:

1078

critical_filter = self.CriticalFilter()

1079

log.addFilter(critical_filter)

1080

try:

1081

with self.assertRaises(SystemExit) as e:

1082

get_managed_objects(MockObjectManagerFailing())

1083

finally:

1084

log.removeFilter(critical_filter)

1085

self.assertTrue(critical_filter.found)

1086

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

1087

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

1088

else:

1089

self.assertIsNotNone(e.exception.code)

1090

1091

class CriticalFilter(logging.Filter):

1092

"""Don't show, but register, critical messages"""

1093

found = False

1094

def filter(self, record):

1095

is_critical = record.levelno >= logging.CRITICAL

1096

self.found = is_critical or self.found

1097

return not is_critical

1098

1099

1100

class Test_SilenceLogger(unittest.TestCase):

1101

loggername = "mandos-ctl.Test_SilenceLogger"

1102

log = logging.getLogger(loggername)

1103

log.propagate = False

1104

log.addHandler(logging.NullHandler())

1105

1106

def setUp(self):

1107

self.counting_filter = self.CountingFilter()

1108

1109

class CountingFilter(logging.Filter):

1110

"Count number of records"

1111

count = 0

1112

def filter(self, record):

1113

self.count += 1

1114

return True

1115

1116

def test_should_filter_records_only_when_active(self):

1117

try:

1118

with SilenceLogger(self.loggername):

1119

self.log.addFilter(self.counting_filter)

1120

self.log.info("Filtered log message 1")

1121

self.log.info("Non-filtered message 2")

1122

self.log.info("Non-filtered message 3")

1123

finally:

1124

self.log.removeFilter(self.counting_filter)

1125

self.assertEqual(self.counting_filter.count, 2)

1126

1127

1128

class Test_commands_from_options(unittest.TestCase):

1129

def setUp(self):

1130

self.parser = argparse.ArgumentParser()

1131

add_command_line_options(self.parser)

1132

1133

def test_is_enabled(self):

1134

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

1135

IsEnabledCmd)

1136

1137

def assert_command_from_args(self, args, command_cls,

1138

**cmd_attrs):

1139

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

1140

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

1141

options = self.parser.parse_args(args)

1142

check_option_syntax(self.parser, options)

1143

commands = commands_from_options(options)

1144

self.assertEqual(len(commands), 1)

1145

command = commands[0]

1146

self.assertIsInstance(command, command_cls)

1147

for key, value in cmd_attrs.items():

1148

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

1149

1150

def test_is_enabled_short(self):

1151

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

1152

1153

def test_approve(self):

1154

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

1155

ApproveCmd)

1156

1157

def test_approve_short(self):

1158

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

1159

1160

def test_deny(self):

1161

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

1162

1163

def test_deny_short(self):

1164

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

1165

1166

def test_remove(self):

1167

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

1168

RemoveCmd)

1169

1170

def test_deny_before_remove(self):

1171

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

1172

"foo"])

1173

check_option_syntax(self.parser, options)

1174

commands = commands_from_options(options)

1175

self.assertEqual(len(commands), 2)

1176

self.assertIsInstance(commands[0], DenyCmd)

1177

self.assertIsInstance(commands[1], RemoveCmd)

1178

1179

def test_deny_before_remove_reversed(self):

1180

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

1181

"--all"])

1182

check_option_syntax(self.parser, options)

1183

commands = commands_from_options(options)

1184

self.assertEqual(len(commands), 2)

1185

self.assertIsInstance(commands[0], DenyCmd)

1186

self.assertIsInstance(commands[1], RemoveCmd)

1187

1188

def test_remove_short(self):

1189

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

1190

1191

def test_dump_json(self):

1192

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

1193

1194

def test_enable(self):

1195

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

1196

1197

def test_enable_short(self):

1198

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

1199

1200

def test_disable(self):

1201

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

1202

DisableCmd)

1203

1204

def test_disable_short(self):

1205

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

1206

1207

def test_bump_timeout(self):

1208

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

1209

BumpTimeoutCmd)

1210

1211

def test_bump_timeout_short(self):

1212

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

1213

1214

def test_start_checker(self):

1215

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

1216

StartCheckerCmd)

1217

1218

def test_stop_checker(self):

1219

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

1220

StopCheckerCmd)

1221

1222

def test_approve_by_default(self):

1223

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

1224

ApproveByDefaultCmd)

1225

1226

def test_deny_by_default(self):

1227

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

1228

DenyByDefaultCmd)

1229

1230

def test_checker(self):

1231

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

1232

SetCheckerCmd, value_to_set=":")

1233

1234

def test_checker_empty(self):

1235

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

1236

SetCheckerCmd, value_to_set="")

1237

1238

def test_checker_short(self):

1239

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

1240

SetCheckerCmd, value_to_set=":")

1241

1242

def test_host(self):

1243

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

1244

"foo"], SetHostCmd,

1245

value_to_set="foo.example.org")

1246

1247

def test_host_short(self):

1248

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

1249

"foo"], SetHostCmd,

1250

value_to_set="foo.example.org")

1251

1252

def test_secret_devnull(self):

1253

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

1254

"foo"], SetSecretCmd,

1255

value_to_set=b"")

1256

1257

def test_secret_tempfile(self):

1258

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

1259

value = b"secret\0xyzzy\nbar"

1260

f.write(value)

1261

f.seek(0)

1262

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

1263

"foo"], SetSecretCmd,

1264

value_to_set=value)

1265

1266

def test_secret_devnull_short(self):

1267

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

1268

SetSecretCmd, value_to_set=b"")

1269

1270

def test_secret_tempfile_short(self):

1271

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

1272

value = b"secret\0xyzzy\nbar"

1273

f.write(value)

1274

f.seek(0)

1275

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

1276

SetSecretCmd,

1277

value_to_set=value)

1278

1279

def test_timeout(self):

1280

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

1281

SetTimeoutCmd,

1282

value_to_set=300000)

1283

1284

def test_timeout_short(self):

1285

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

1286

SetTimeoutCmd,

1287

value_to_set=300000)

1288

1289

def test_extended_timeout(self):

1290

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

1291

"foo"],

1292

SetExtendedTimeoutCmd,

1293

value_to_set=900000)

1294

1295

def test_interval(self):

1296

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

1297

SetIntervalCmd,

1298

value_to_set=120000)

1299

1300

def test_interval_short(self):

1301

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

1302

SetIntervalCmd,

1303

value_to_set=120000)

1304

1305

def test_approval_delay(self):

1306

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

1307

"foo"], SetApprovalDelayCmd,

1308

value_to_set=30000)

1309

1310

def test_approval_duration(self):

1311

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

1312

"foo"], SetApprovalDurationCmd,

1313

value_to_set=1000)

1314

1315

def test_print_table(self):

1316

self.assert_command_from_args([], PrintTableCmd,

1317

verbose=False)

1318

1319

def test_print_table_verbose(self):

1320

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

1321

verbose=True)

1322

1323

def test_print_table_verbose_short(self):

1324

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

1325

verbose=True)

1326

1327

1328

class TestCmd(unittest.TestCase):

1329

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

1330

1331

def setUp(self):

1332

testcase = self

1333

class MockClient(object):

1334

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

1335

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

1336

self.attributes = attributes

1337

self.attributes["Name"] = name

1338

self.calls = []

1339

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

1340

testcase.assertEqual(interface, client_dbus_interface)

1341

testcase.assertEqual(dbus_interface,

1342

dbus.PROPERTIES_IFACE)

1343

self.attributes[propname] = value

1344

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

1345

testcase.assertEqual(interface, client_dbus_interface)

1346

testcase.assertEqual(dbus_interface,

1347

dbus.PROPERTIES_IFACE)

1348

return self.attributes[propname]

1349

def Approve(self, approve, dbus_interface):

1350

testcase.assertEqual(dbus_interface,

1351

client_dbus_interface)

1352

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

1353

dbus_interface)))

1354

self.client = MockClient(

1355

"foo",

1356

KeyID=("92ed150794387c03ce684574b1139a65"

1357

"94a34f895daaaf09fd8ea90a27cddb12"),

1358

Secret=b"secret",

1359

Host="foo.example.org",

1360

Enabled=dbus.Boolean(True),

1361

Timeout=300000,

1362

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

1363

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

1364

Interval=120000,

1365

Fingerprint=("778827225BA7DE539C5A"

1366

"7CFA59CFF7CDBD9A5920"),

1367

CheckerRunning=dbus.Boolean(False),

1368

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

1369

ApprovalPending=dbus.Boolean(False),

1370

ApprovedByDefault=dbus.Boolean(True),

1371

LastApprovalRequest="",

1372

ApprovalDelay=0,

1373

ApprovalDuration=1000,

1374

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

1375

ExtendedTimeout=900000,

1376

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

1377

LastCheckerStatus=0)

1378

self.other_client = MockClient(

1379

"barbar",

1380

KeyID=("0558568eedd67d622f5c83b35a115f79"

1381

"6ab612cff5ad227247e46c2b020f441c"),

1382

Secret=b"secretbar",

1383

Host="192.0.2.3",

1384

Enabled=dbus.Boolean(True),

1385

Timeout=300000,

1386

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

1387

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

1388

Interval=120000,

1389

Fingerprint=("3E393AEAEFB84C7E89E2"

1390

"F547B3A107558FCA3A27"),

1391

CheckerRunning=dbus.Boolean(True),

1392

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

1393

ApprovalPending=dbus.Boolean(False),

1394

ApprovedByDefault=dbus.Boolean(False),

1395

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

1396

ApprovalDelay=30000,

1397

ApprovalDuration=93785000,

1398

Checker=":",

1399

ExtendedTimeout=900000,

1400

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

1401

LastCheckerStatus=-2)

1402

self.clients = collections.OrderedDict(

1403

[

1404

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

1405

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

1406

])

1407

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

1408

1409

@property

1410

def bus(self):

1411

class Bus(object):

1412

@staticmethod

1413

def get_object(client_bus_name, path):

1414

self.assertEqual(client_bus_name, dbus_busname)

1415

return {

1416

# Note: "self" here is the TestCmd instance, not

1417

# the Bus instance, since this is a static method!

1418

"/clients/foo": self.client,

1419

"/clients/barbar": self.other_client,

1420

}[path]

1421

return Bus()

1422

1423

1424

class TestIsEnabledCmd(TestCmd):

1425

def test_is_enabled(self):

1426

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

1427

properties)

1428

for client, properties

1429

in self.clients.items()))

1430

1431

def test_is_enabled_run_exits_successfully(self):

1432

with self.assertRaises(SystemExit) as e:

1433

IsEnabledCmd().run(self.one_client)

1434

if e.exception.code is not None:

1435

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

1436

else:

1437

self.assertIsNone(e.exception.code)

1438

1439

def test_is_enabled_run_exits_with_failure(self):

1440

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

1441

with self.assertRaises(SystemExit) as e:

1442

IsEnabledCmd().run(self.one_client)

1443

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

1444

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

1445

else:

1446

self.assertIsNotNone(e.exception.code)

1447

1448

1449

class TestApproveCmd(TestCmd):

1450

def test_approve(self):

1451

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

1452

for clientpath in self.clients:

1453

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

1454

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

1455

client.calls)

1456

1457

1458

class TestDenyCmd(TestCmd):

1459

def test_deny(self):

1460

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

1461

for clientpath in self.clients:

1462

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

1463

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

1464

client.calls)

1465

1466

1467

class TestRemoveCmd(TestCmd):

1468

def test_remove(self):

1469

class MockMandos(object):

1470

def __init__(self):

1471

self.calls = []

1472

def RemoveClient(self, dbus_path):

1473

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

1474

mandos = MockMandos()

1475

super(TestRemoveCmd, self).setUp()

1476

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

1477

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

1478

for clientpath in self.clients:

1479

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

1480

mandos.calls)

1481

1482

1483

class TestDumpJSONCmd(TestCmd):

1484

def setUp(self):

1485

self.expected_json = {

1486

"foo": {

1487

"Name": "foo",

1488

"KeyID": ("92ed150794387c03ce684574b1139a65"

1489

"94a34f895daaaf09fd8ea90a27cddb12"),

1490

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

1491

"Enabled": True,

1492

"Timeout": 300000,

1493

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

1494

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

1495

"Interval": 120000,

1496

"Fingerprint": ("778827225BA7DE539C5A"

1497

"7CFA59CFF7CDBD9A5920"),

1498

"CheckerRunning": False,

1499

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

1500

"ApprovalPending": False,

1501

"ApprovedByDefault": True,

1502

"LastApprovalRequest": "",

1503

"ApprovalDelay": 0,

1504

"ApprovalDuration": 1000,

1505

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

1506

"ExtendedTimeout": 900000,

1507

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

1508

"LastCheckerStatus": 0,

1509

},

1510

"barbar": {

1511

"Name": "barbar",

1512

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1513

"6ab612cff5ad227247e46c2b020f441c"),

1514

"Host": "192.0.2.3",

1515

"Enabled": True,

1516

"Timeout": 300000,

1517

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

1518

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

1519

"Interval": 120000,

1520

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1521

"F547B3A107558FCA3A27"),

1522

"CheckerRunning": True,

1523

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

1524

"ApprovalPending": False,

1525

"ApprovedByDefault": False,

1526

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

1527

"ApprovalDelay": 30000,

1528

"ApprovalDuration": 93785000,

1529

"Checker": ":",

1530

"ExtendedTimeout": 900000,

1531

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

1532

"LastCheckerStatus": -2,

1533

},

1534

}

1535

return super(TestDumpJSONCmd, self).setUp()

1536

1537

def test_normal(self):

1538

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

1539

json_data = json.loads(output)

1540

self.assertDictEqual(json_data, self.expected_json)

1541

1542

def test_one_client(self):

1543

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

1544

json_data = json.loads(output)

1545

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

1546

self.assertDictEqual(json_data, expected_json)

1547

1548

1549

class TestPrintTableCmd(TestCmd):

1550

def test_normal(self):

1551

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

1552

expected_output = "\n".join((

1553

"Name Enabled Timeout Last Successful Check",

1554

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

1555

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

1556

))

1557

self.assertEqual(output, expected_output)

1558

1559

def test_verbose(self):

1560

output = PrintTableCmd(verbose=True).output(

1561

self.clients.values())

1562

columns = (

1563

(

1564

"Name ",

1565

"foo ",

1566

"barbar ",

1567

),(

1568

"Enabled ",

1569

"Yes ",

1570

"Yes ",

1571

),(

1572

"Timeout ",

1573

"00:05:00 ",

1574

"00:05:00 ",

1575

),(

1576

"Last Successful Check ",

1577

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

1578

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

1579

),(

1580

"Created ",

1581

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

1582

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

1583

),(

1584

"Interval ",

1585

"00:02:00 ",

1586

"00:02:00 ",

1587

),(

1588

"Host ",

1589

"foo.example.org ",

1590

"192.0.2.3 ",

1591

),(

1592

("Key ID "

1593

" "),

1594

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1595

"ea90a27cddb12 "),

1596

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1597

"46c2b020f441c "),

1598

),(

1599

"Fingerprint ",

1600

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1601

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1602

),(

1603

"Check Is Running ",

1604

"No ",

1605

"Yes ",

1606

),(

1607

"Last Enabled ",

1608

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

1609

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

1610

),(

1611

"Approval Is Pending ",

1612

"No ",

1613

"No ",

1614

),(

1615

"Approved By Default ",

1616

"Yes ",

1617

"No ",

1618

),(

1619

"Last Approval Request ",

1620

" ",

1621

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

1622

),(

1623

"Approval Delay ",

1624

"00:00:00 ",

1625

"00:00:30 ",

1626

),(

1627

"Approval Duration ",

1628

"00:00:01 ",

1629

"1T02:03:05 ",

1630

),(

1631

"Checker ",

1632

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

1633

": ",

1634

),(

1635

"Extended Timeout ",

1636

"00:15:00 ",

1637

"00:15:00 ",

1638

),(

1639

"Expires ",

1640

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

1641

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

1642

),(

1643

"Last Checker Status",

1644

"0 ",

1645

"-2 ",

1646

)

1647

)

1648

num_lines = max(len(rows) for rows in columns)

1649

expected_output = "\n".join("".join(rows[line]

1650

for rows in columns)

1651

for line in range(num_lines))

1652

self.assertEqual(output, expected_output)

1653

1654

def test_one_client(self):

1655

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

1656

expected_output = "\n".join((

1657

"Name Enabled Timeout Last Successful Check",

1658

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

1659

))

1660

self.assertEqual(output, expected_output)

1661

1662

1663

class TestPropertyCmd(TestCmd):

1664

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

1665

def runTest(self):

1666

if not hasattr(self, "command"):

1667

return

1668

values_to_get = getattr(self, "values_to_get",

1669

self.values_to_set)

1670

for value_to_set, value_to_get in zip(self.values_to_set,

1671

values_to_get):

1672

for clientpath in self.clients:

1673

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

1674

old_value = client.attributes[self.propname]

1675

self.assertNotIsInstance(old_value, self.Unique)

1676

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

1677

self.run_command(value_to_set, self.clients)

1678

for clientpath in self.clients:

1679

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

1680

value = client.attributes[self.propname]

1681

self.assertNotIsInstance(value, self.Unique)

1682

self.assertEqual(value, value_to_get)

1683

1684

class Unique(object):

1685

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

1686

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

1687

1688

def run_command(self, value, clients):

1689

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

1690

1691

1692

class TestEnableCmd(TestPropertyCmd):

1693

command = EnableCmd

1694

propname = "Enabled"

1695

values_to_set = [dbus.Boolean(True)]

1696

1697

1698

class TestDisableCmd(TestPropertyCmd):

1699

command = DisableCmd

1700

propname = "Enabled"

1701

values_to_set = [dbus.Boolean(False)]

1702

1703

1704

class TestBumpTimeoutCmd(TestPropertyCmd):

1705

command = BumpTimeoutCmd

1706

propname = "LastCheckedOK"

1707

values_to_set = [""]

1708

1709

1710

class TestStartCheckerCmd(TestPropertyCmd):

1711

command = StartCheckerCmd

1712

propname = "CheckerRunning"

1713

values_to_set = [dbus.Boolean(True)]

1714

1715

1716

class TestStopCheckerCmd(TestPropertyCmd):

1717

command = StopCheckerCmd

1718

propname = "CheckerRunning"

1719

values_to_set = [dbus.Boolean(False)]

1720

1721

1722

class TestApproveByDefaultCmd(TestPropertyCmd):

1723

command = ApproveByDefaultCmd

1724

propname = "ApprovedByDefault"

1725

values_to_set = [dbus.Boolean(True)]

1726

1727

1728

class TestDenyByDefaultCmd(TestPropertyCmd):

1729

command = DenyByDefaultCmd

1730

propname = "ApprovedByDefault"

1731

values_to_set = [dbus.Boolean(False)]

1732

1733

1734

class TestPropertyValueCmd(TestPropertyCmd):

1735

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

1736

1737

def runTest(self):

1738

if type(self) is TestPropertyValueCmd:

1739

return

1740

return super(TestPropertyValueCmd, self).runTest()

1741

1742

def run_command(self, value, clients):

1743

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

1744

1745

1746

class TestSetCheckerCmd(TestPropertyValueCmd):

1747

command = SetCheckerCmd

1748

propname = "Checker"

1749

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

1750

1751

1752

class TestSetHostCmd(TestPropertyValueCmd):

1753

command = SetHostCmd

1754

propname = "Host"

1755

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

1756

1757

1758

class TestSetSecretCmd(TestPropertyValueCmd):

1759

command = SetSecretCmd

1760

propname = "Secret"

1761

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

1762

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

1763

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

1764

1765

1766

class TestSetTimeoutCmd(TestPropertyValueCmd):

1767

command = SetTimeoutCmd

1768

propname = "Timeout"

1769

values_to_set = [datetime.timedelta(),

1770

datetime.timedelta(minutes=5),

1771

datetime.timedelta(seconds=1),

1772

datetime.timedelta(weeks=1),

1773

datetime.timedelta(weeks=52)]

1774

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

1775

1776

1777

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1778

command = SetExtendedTimeoutCmd

1779

propname = "ExtendedTimeout"

1780

values_to_set = [datetime.timedelta(),

1781

datetime.timedelta(minutes=5),

1782

datetime.timedelta(seconds=1),

1783

datetime.timedelta(weeks=1),

1784

datetime.timedelta(weeks=52)]

1785

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

1786

1787

1788

class TestSetIntervalCmd(TestPropertyValueCmd):

1789

command = SetIntervalCmd

1790

propname = "Interval"

1791

values_to_set = [datetime.timedelta(),

1792

datetime.timedelta(minutes=5),

1793

datetime.timedelta(seconds=1),

1794

datetime.timedelta(weeks=1),

1795

datetime.timedelta(weeks=52)]

1796

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

1797

1798

1799

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1800

command = SetApprovalDelayCmd

1801

propname = "ApprovalDelay"

1802

values_to_set = [datetime.timedelta(),

1803

datetime.timedelta(minutes=5),

1804

datetime.timedelta(seconds=1),

1805

datetime.timedelta(weeks=1),

1806

datetime.timedelta(weeks=52)]

1807

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

1808

1809

1810

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1811

command = SetApprovalDurationCmd

1812

propname = "ApprovalDuration"

1813

values_to_set = [datetime.timedelta(),

1814

datetime.timedelta(minutes=5),

1815

datetime.timedelta(seconds=1),

1816

datetime.timedelta(weeks=1),

1817

datetime.timedelta(weeks=52)]

1818

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

1819

1820

1821

1822

def should_only_run_tests():

1823

parser = argparse.ArgumentParser(add_help=False)

1824

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

1825

args, unknown_args = parser.parse_known_args()

1826

run_tests = args.check

1827

if run_tests:

1828

# Remove --check argument from sys.argv

1829

sys.argv[1:] = unknown_args

1830

return run_tests

1831

1832

# Add all tests from doctest strings

1833

def load_tests(loader, tests, none):

1834

import doctest

1835

tests.addTests(doctest.DocTestSuite())

1836

return tests

341

1837

342

1838

if __name__ == "__main__":

343

main()

1839

try:

1840

if should_only_run_tests():

1841

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

1842

unittest.main()

1843

else:

1844

main()

1845

finally:

1846

logging.shutdown()