/mandos/release : revision 237.7.621

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 <http://www.gnu.org/licenses/>.

21

#

22

# Contact the authors at <mandos@fukt.bsnet.se>.

23

#

22

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

23

#

24

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

25

#

24

26

25

27

from __future__ import (division, absolute_import, print_function,

26

28

unicode_literals)

27

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

28

35

import sys

29

import dbus

30

36

import argparse

31

37

import locale

32

38

import datetime

33

39

import re

34

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

35

64

36

65

locale.setlocale(locale.LC_ALL, "")

37

66

38

tablewords = {

39

"Name": "Name",

40

"Enabled": "Enabled",

41

"Timeout": "Timeout",

42

"LastCheckedOK": "Last Successful Check",

43

"LastApprovalRequest": "Last Approval Request",

44

"Created": "Created",

45

"Interval": "Interval",

46

"Host": "Host",

47

"Fingerprint": "Fingerprint",

48

"CheckerRunning": "Check Is Running",

49

"LastEnabled": "Last Enabled",

50

"ApprovalPending": "Approval Is Pending",

51

"ApprovedByDefault": "Approved By Default",

52

"ApprovalDelay": "Approval Delay",

53

"ApprovalDuration": "Approval Duration",

54

"Checker": "Checker",

55

}

56

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

57

domain = "se.bsnet.fukt"

58

busname = domain + ".Mandos"

59

server_path = "/"

60

server_interface = domain + ".Mandos"

61

client_interface = domain + ".Mandos.Client"

62

version = "1.3.0"

63

64

def timedelta_to_milliseconds(td):

65

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

66

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

67

+ (td.seconds * 1000)

68

+ (td.microseconds // 1000))

69

70

def milliseconds_to_string(ms):

71

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

72

return ("%(days)s%(hours)02d:%(minutes)02d:%(seconds)02d"

73

% { "days": "%dT" % td.days if td.days else "",

74

"hours": td.seconds // 3600,

75

"minutes": (td.seconds % 3600) // 60,

76

"seconds": td.seconds % 60,

77

})

78

79

def string_to_delta(interval):

80

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

81

82

>>> string_to_delta("7d")

83

datetime.timedelta(7)

84

>>> string_to_delta("60s")

85

datetime.timedelta(0, 60)

86

>>> string_to_delta("60m")

87

datetime.timedelta(0, 3600)

88

>>> string_to_delta("24h")

89

datetime.timedelta(1)

90

>>> string_to_delta("1w")

91

datetime.timedelta(7)

92

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

93

datetime.timedelta(0, 330)

94

"""

95

timevalue = datetime.timedelta(0)

96

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

97

98

for s in regexp.findall(interval):

99

try:

100

suffix = unicode(s[-1])

101

value = int(s[:-1])

102

if suffix == "d":

103

delta = datetime.timedelta(value)

104

elif suffix == "s":

105

delta = datetime.timedelta(0, value)

106

elif suffix == "m":

107

delta = datetime.timedelta(0, 0, 0, 0, value)

108

elif suffix == "h":

109

delta = datetime.timedelta(0, 0, 0, 0, 0, value)

110

elif suffix == "w":

111

delta = datetime.timedelta(0, 0, 0, 0, 0, 0, value)

112

else:

113

raise ValueError

114

except (ValueError, IndexError):

115

raise ValueError

116

timevalue += delta

117

return timevalue

118

119

def print_clients(clients, keywords):

120

def valuetostring(value, keyword):

121

if type(value) is dbus.Boolean:

122

return "Yes" if value else "No"

123

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

124

"ApprovalDuration"):

125

return milliseconds_to_string(value)

126

return unicode(value)

127

128

# Create format string to print table rows

129

format_string = " ".join("%%-%ds" %

130

max(len(tablewords[key]),

131

max(len(valuetostring(client[key],

132

key))

133

for client in

134

clients))

135

for key in keywords)

136

# Print header line

137

print(format_string % tuple(tablewords[key] for key in keywords))

138

for client in clients:

139

print(format_string % tuple(valuetostring(client[key], key)

140

for key in keywords))

141

142

def has_actions(options):

143

return any((options.enable,

144

options.disable,

145

options.bump_timeout,

146

options.start_checker,

147

options.stop_checker,

148

options.is_enabled,

149

options.remove,

150

options.checker is not None,

151

options.timeout is not None,

152

options.interval is not None,

153

options.approved_by_default is not None,

154

options.approval_delay is not None,

155

options.approval_duration is not None,

156

options.host is not None,

157

options.secret is not None,

158

options.approve,

159

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

160

81

161

82

def main():

162

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):

163

135

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

164

version = "%%prog %s" % version,

136

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

165

137

help="show version number and exit")

166

138

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

167

139

help="Select all clients")

168

140

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

169

141

help="Print all fields")

170

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

171

help="Enable client")

172

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

173

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

174

150

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

175

151

help="Bump timeout for client")

176

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

177

help="Start checker for client")

178

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

179

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

180

159

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

181

160

help="Check if client is enabled")

182

161

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

183

162

help="Remove client")

184

163

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

185

164

help="Set checker command for client")

186

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

165

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

187

166

help="Set timeout for client")

188

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

167

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

168

help="Set extended timeout for client")

169

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

189

170

help="Set checker interval for client")

190

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

191

default=None, dest="approved_by_default",

192

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

193

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

194

dest="approved_by_default",

195

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

196

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,

197

181

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

198

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

182

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

199

183

help="Set duration of one client approval")

200

184

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

201

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

185

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

186

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

202

187

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

203

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

204

help="Approve any current client request")

205

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

206

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

207

198

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

208

options = parser.parse_args()

209

210

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

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

409

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

211

410

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

212

411

if options.verbose and has_actions(options):

213

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

214

" --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.")

215

416

if options.all and not has_actions(options):

216

417

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

217

218

try:

219

bus = dbus.SystemBus()

220

mandos_dbus_objc = bus.get_object(busname, server_path)

221

except dbus.exceptions.DBusException:

222

print("Could not connect to Mandos server",

223

file=sys.stderr)

224

sys.exit(1)

225

226

mandos_serv = dbus.Interface(mandos_dbus_objc,

227

dbus_interface = server_interface)

228

229

#block stderr since dbus library prints to stderr

230

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

231

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

232

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

233

os.close(null)

234

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 TestCaseWithAssertLogs(unittest.TestCase):

843

"""unittest.TestCase.assertLogs only exists in Python 3.4"""

844

845

if not hasattr(unittest.TestCase, "assertLogs"):

846

@contextlib.contextmanager

847

def assertLogs(self, logger, level=logging.INFO):

848

capturing_handler = self.CapturingLevelHandler(level)

849

old_level = logger.level

850

old_propagate = logger.propagate

851

logger.addHandler(capturing_handler)

852

logger.setLevel(level)

853

logger.propagate = False

854

try:

855

yield capturing_handler.watcher

856

finally:

857

logger.propagate = old_propagate

858

logger.removeHandler(capturing_handler)

859

logger.setLevel(old_level)

860

self.assertGreater(len(capturing_handler.watcher.records),

861

0)

862

863

class CapturingLevelHandler(logging.Handler):

864

def __init__(self, level, *args, **kwargs):

865

logging.Handler.__init__(self, *args, **kwargs)

866

self.watcher = self.LoggingWatcher([], [])

867

def emit(self, record):

868

self.watcher.records.append(record)

869

self.watcher.output.append(self.format(record))

870

871

LoggingWatcher = collections.namedtuple("LoggingWatcher",

872

("records",

873

"output"))

874

875

class Test_string_to_delta(TestCaseWithAssertLogs):

876

def test_handles_basic_rfc3339(self):

877

self.assertEqual(string_to_delta("PT0S"),

878

datetime.timedelta())

879

self.assertEqual(string_to_delta("P0D"),

880

datetime.timedelta())

881

self.assertEqual(string_to_delta("PT1S"),

882

datetime.timedelta(0, 1))

883

self.assertEqual(string_to_delta("PT2H"),

884

datetime.timedelta(0, 7200))

885

886

def test_falls_back_to_pre_1_6_1_with_warning(self):

887

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

888

value = string_to_delta("2h")

889

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

890

891

892

class Test_check_option_syntax(unittest.TestCase):

893

def setUp(self):

894

self.parser = argparse.ArgumentParser()

895

add_command_line_options(self.parser)

896

897

def test_actions_requires_client_or_all(self):

898

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

899

options = self.parser.parse_args()

900

setattr(options, action, value)

901

with self.assertParseError():

902

self.check_option_syntax(options)

903

904

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

905

# check_option_syntax()

906

actions = {

907

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

908

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

909

# never used

910

"enable": True,

911

"disable": True,

912

"bump_timeout": True,

913

"start_checker": True,

914

"stop_checker": True,

915

"is_enabled": True,

916

"remove": True,

917

"checker": "x",

918

"timeout": datetime.timedelta(),

919

"extended_timeout": datetime.timedelta(),

920

"interval": datetime.timedelta(),

921

"approved_by_default": True,

922

"approval_delay": datetime.timedelta(),

923

"approval_duration": datetime.timedelta(),

924

"host": "x",

925

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

926

"approve": True,

927

"deny": True,

928

}

929

930

@contextlib.contextmanager

931

def assertParseError(self):

932

with self.assertRaises(SystemExit) as e:

933

with self.temporarily_suppress_stderr():

934

yield

935

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

936

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

937

# /argparse.html#exiting-methods

938

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

939

940

@staticmethod

941

@contextlib.contextmanager

942

def temporarily_suppress_stderr():

943

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

944

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

945

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

946

os.close(null)

235

947

try:

236

mandos_clients = mandos_serv.GetAllClientsWithProperties()

948

yield

237

949

finally:

238

#restore stderr

950

# restore stderr

239

951

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

240

952

os.close(stderrcopy)

241

except dbus.exceptions.DBusException, e:

242

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

243

file=sys.stderr)

244

sys.exit(1)

245

246

# Compile dict of (clients: properties) to process

247

clients={}

248

249

if options.all or not options.client:

250

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

251

for path, properties in

252

mandos_clients.iteritems())

253

else:

254

for name in options.client:

255

for path, client in mandos_clients.iteritems():

256

if client["Name"] == name:

257

client_objc = bus.get_object(busname, path)

258

clients[client_objc] = client

259

break

260

else:

261

print("Client not found on server: %r" % name,

262

file=sys.stderr)

263

sys.exit(1)

264

265

if not has_actions(options) and clients:

266

if options.verbose:

267

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

268

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

269

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

270

"LastEnabled", "ApprovalPending",

271

"ApprovedByDefault",

272

"LastApprovalRequest", "ApprovalDelay",

273

"ApprovalDuration", "Checker")

274

else:

275

keywords = defaultkeywords

276

277

print_clients(clients.values(), keywords)

278

else:

279

# Process each client in the list by all selected options

280

for client in clients:

281

if options.remove:

282

mandos_serv.RemoveClient(client.__dbus_object_path__)

283

if options.enable:

284

client.Enable(dbus_interface=client_interface)

285

if options.disable:

286

client.Disable(dbus_interface=client_interface)

287

if options.bump_timeout:

288

client.CheckedOK(dbus_interface=client_interface)

289

if options.start_checker:

290

client.StartChecker(dbus_interface=client_interface)

291

if options.stop_checker:

292

client.StopChecker(dbus_interface=client_interface)

293

if options.is_enabled:

294

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

295

"Enabled",

296

dbus_interface=

297

dbus.PROPERTIES_IFACE)

298

else 1)

299

if options.checker:

300

client.Set(client_interface, "Checker",

301

options.checker,

302

dbus_interface=dbus.PROPERTIES_IFACE)

303

if options.host:

304

client.Set(client_interface, "Host", options.host,

305

dbus_interface=dbus.PROPERTIES_IFACE)

306

if options.interval:

307

client.Set(client_interface, "Interval",

308

timedelta_to_milliseconds

309

(string_to_delta(options.interval)),

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

if options.approval_delay:

312

client.Set(client_interface, "ApprovalDelay",

313

timedelta_to_milliseconds

314

(string_to_delta(options.

315

approval_delay)),

316

dbus_interface=dbus.PROPERTIES_IFACE)

317

if options.approval_duration:

318

client.Set(client_interface, "ApprovalDuration",

319

timedelta_to_milliseconds

320

(string_to_delta(options.

321

approval_duration)),

322

dbus_interface=dbus.PROPERTIES_IFACE)

323

if options.timeout:

324

client.Set(client_interface, "Timeout",

325

timedelta_to_milliseconds

326

(string_to_delta(options.timeout)),

327

dbus_interface=dbus.PROPERTIES_IFACE)

328

if options.secret:

329

client.Set(client_interface, "Secret",

330

dbus.ByteArray(open(options.secret,

331

"rb").read()),

332

dbus_interface=dbus.PROPERTIES_IFACE)

333

if options.approved_by_default is not None:

334

client.Set(client_interface, "ApprovedByDefault",

335

dbus.Boolean(options

336

.approved_by_default),

337

dbus_interface=dbus.PROPERTIES_IFACE)

338

if options.approve:

339

client.Approve(dbus.Boolean(True),

340

dbus_interface=client_interface)

341

elif options.deny:

342

client.Approve(dbus.Boolean(False),

343

dbus_interface=client_interface)

953

954

def check_option_syntax(self, options):

955

check_option_syntax(self.parser, options)

956

957

def test_actions_conflicts_with_verbose(self):

958

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

959

options = self.parser.parse_args()

960

setattr(options, action, value)

961

options.verbose = True

962

with self.assertParseError():

963

self.check_option_syntax(options)

964

965

def test_dump_json_conflicts_with_verbose(self):

966

options = self.parser.parse_args()

967

options.dump_json = True

968

options.verbose = True

969

with self.assertParseError():

970

self.check_option_syntax(options)

971

972

def test_dump_json_conflicts_with_action(self):

973

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

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.dump_json = True

977

with self.assertParseError():

978

self.check_option_syntax(options)

979

980

def test_all_can_not_be_alone(self):

981

options = self.parser.parse_args()

982

options.all = True

983

with self.assertParseError():

984

self.check_option_syntax(options)

985

986

def test_all_is_ok_with_any_action(self):

987

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

988

options = self.parser.parse_args()

989

setattr(options, action, value)

990

options.all = True

991

self.check_option_syntax(options)

992

993

def test_is_enabled_fails_without_client(self):

994

options = self.parser.parse_args()

995

options.is_enabled = True

996

with self.assertParseError():

997

self.check_option_syntax(options)

998

999

def test_is_enabled_works_with_one_client(self):

1000

options = self.parser.parse_args()

1001

options.is_enabled = True

1002

options.client = ["foo"]

1003

self.check_option_syntax(options)

1004

1005

def test_is_enabled_fails_with_two_clients(self):

1006

options = self.parser.parse_args()

1007

options.is_enabled = True

1008

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

1009

with self.assertParseError():

1010

self.check_option_syntax(options)

1011

1012

def test_remove_can_only_be_combined_with_action_deny(self):

1013

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

1014

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

1015

continue

1016

options = self.parser.parse_args()

1017

setattr(options, action, value)

1018

options.all = True

1019

options.remove = True

1020

with self.assertParseError():

1021

self.check_option_syntax(options)

1022

1023

1024

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1025

def test_calls_and_returns_get_object_on_bus(self):

1026

class MockBus(object):

1027

called = False

1028

def get_object(mockbus_self, busname, dbus_path):

1029

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

1030

# this MockBus instance is in "mockbus_self".

1031

self.assertEqual(busname, dbus_busname)

1032

self.assertEqual(dbus_path, server_dbus_path)

1033

mockbus_self.called = True

1034

return mockbus_self

1035

1036

mockbus = get_mandos_dbus_object(bus=MockBus())

1037

self.assertIsInstance(mockbus, MockBus)

1038

self.assertTrue(mockbus.called)

1039

1040

def test_logs_and_exits_on_dbus_error(self):

1041

class MockBusFailing(object):

1042

def get_object(self, busname, dbus_path):

1043

raise dbus.exceptions.DBusException("Test")

1044

1045

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

1046

with self.assertRaises(SystemExit) as e:

1047

bus = get_mandos_dbus_object(bus=MockBusFailing())

1048

1049

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

1050

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

1051

else:

1052

self.assertIsNotNone(e.exception.code)

1053

1054

1055

class Test_get_managed_objects(TestCaseWithAssertLogs):

1056

def test_calls_and_returns_GetManagedObjects(self):

1057

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

1058

class MockObjectManager(object):

1059

def GetManagedObjects(self):

1060

return managed_objects

1061

retval = get_managed_objects(MockObjectManager())

1062

self.assertDictEqual(managed_objects, retval)

1063

1064

def test_logs_and_exits_on_dbus_error(self):

1065

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

1066

1067

class MockObjectManagerFailing(object):

1068

def GetManagedObjects(self):

1069

dbus_logger.error("Test")

1070

raise dbus.exceptions.DBusException("Test")

1071

1072

class CountingHandler(logging.Handler):

1073

count = 0

1074

def emit(self, record):

1075

self.count += 1

1076

1077

counting_handler = CountingHandler()

1078

1079

dbus_logger.addHandler(counting_handler)

1080

1081

try:

1082

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

1083

with self.assertRaises(SystemExit) as e:

1084

get_managed_objects(MockObjectManagerFailing())

1085

finally:

1086

dbus_logger.removeFilter(counting_handler)

1087

self.assertEqual(counting_handler.count, 0)

1088

1089

# Test that the dbus_logger still works

1090

with self.assertLogs(dbus_logger, logging.ERROR):

1091

dbus_logger.error("Test")

1092

1093

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

1094

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

1095

else:

1096

self.assertIsNotNone(e.exception.code)

1097

1098

1099

class Test_commands_from_options(unittest.TestCase):

1100

def setUp(self):

1101

self.parser = argparse.ArgumentParser()

1102

add_command_line_options(self.parser)

1103

1104

def test_is_enabled(self):

1105

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

1106

IsEnabledCmd)

1107

1108

def assert_command_from_args(self, args, command_cls,

1109

**cmd_attrs):

1110

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

1111

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

1112

options = self.parser.parse_args(args)

1113

check_option_syntax(self.parser, options)

1114

commands = commands_from_options(options)

1115

self.assertEqual(len(commands), 1)

1116

command = commands[0]

1117

self.assertIsInstance(command, command_cls)

1118

for key, value in cmd_attrs.items():

1119

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

1120

1121

def test_is_enabled_short(self):

1122

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

1123

1124

def test_approve(self):

1125

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

1126

ApproveCmd)

1127

1128

def test_approve_short(self):

1129

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

1130

1131

def test_deny(self):

1132

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

1133

1134

def test_deny_short(self):

1135

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

1136

1137

def test_remove(self):

1138

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

1139

RemoveCmd)

1140

1141

def test_deny_before_remove(self):

1142

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

1143

"foo"])

1144

check_option_syntax(self.parser, options)

1145

commands = commands_from_options(options)

1146

self.assertEqual(len(commands), 2)

1147

self.assertIsInstance(commands[0], DenyCmd)

1148

self.assertIsInstance(commands[1], RemoveCmd)

1149

1150

def test_deny_before_remove_reversed(self):

1151

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

1152

"--all"])

1153

check_option_syntax(self.parser, options)

1154

commands = commands_from_options(options)

1155

self.assertEqual(len(commands), 2)

1156

self.assertIsInstance(commands[0], DenyCmd)

1157

self.assertIsInstance(commands[1], RemoveCmd)

1158

1159

def test_remove_short(self):

1160

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

1161

1162

def test_dump_json(self):

1163

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

1164

1165

def test_enable(self):

1166

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

1167

1168

def test_enable_short(self):

1169

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

1170

1171

def test_disable(self):

1172

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

1173

DisableCmd)

1174

1175

def test_disable_short(self):

1176

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

1177

1178

def test_bump_timeout(self):

1179

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

1180

BumpTimeoutCmd)

1181

1182

def test_bump_timeout_short(self):

1183

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

1184

1185

def test_start_checker(self):

1186

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

1187

StartCheckerCmd)

1188

1189

def test_stop_checker(self):

1190

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

1191

StopCheckerCmd)

1192

1193

def test_approve_by_default(self):

1194

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

1195

ApproveByDefaultCmd)

1196

1197

def test_deny_by_default(self):

1198

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

1199

DenyByDefaultCmd)

1200

1201

def test_checker(self):

1202

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

1203

SetCheckerCmd, value_to_set=":")

1204

1205

def test_checker_empty(self):

1206

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

1207

SetCheckerCmd, value_to_set="")

1208

1209

def test_checker_short(self):

1210

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

1211

SetCheckerCmd, value_to_set=":")

1212

1213

def test_host(self):

1214

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

1215

"foo"], SetHostCmd,

1216

value_to_set="foo.example.org")

1217

1218

def test_host_short(self):

1219

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

1220

"foo"], SetHostCmd,

1221

value_to_set="foo.example.org")

1222

1223

def test_secret_devnull(self):

1224

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

1225

"foo"], SetSecretCmd,

1226

value_to_set=b"")

1227

1228

def test_secret_tempfile(self):

1229

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

1230

value = b"secret\0xyzzy\nbar"

1231

f.write(value)

1232

f.seek(0)

1233

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

1234

"foo"], SetSecretCmd,

1235

value_to_set=value)

1236

1237

def test_secret_devnull_short(self):

1238

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

1239

SetSecretCmd, value_to_set=b"")

1240

1241

def test_secret_tempfile_short(self):

1242

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

1243

value = b"secret\0xyzzy\nbar"

1244

f.write(value)

1245

f.seek(0)

1246

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

1247

SetSecretCmd,

1248

value_to_set=value)

1249

1250

def test_timeout(self):

1251

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

1252

SetTimeoutCmd,

1253

value_to_set=300000)

1254

1255

def test_timeout_short(self):

1256

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

1257

SetTimeoutCmd,

1258

value_to_set=300000)

1259

1260

def test_extended_timeout(self):

1261

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

1262

"foo"],

1263

SetExtendedTimeoutCmd,

1264

value_to_set=900000)

1265

1266

def test_interval(self):

1267

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

1268

SetIntervalCmd,

1269

value_to_set=120000)

1270

1271

def test_interval_short(self):

1272

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

1273

SetIntervalCmd,

1274

value_to_set=120000)

1275

1276

def test_approval_delay(self):

1277

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

1278

"foo"], SetApprovalDelayCmd,

1279

value_to_set=30000)

1280

1281

def test_approval_duration(self):

1282

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

1283

"foo"], SetApprovalDurationCmd,

1284

value_to_set=1000)

1285

1286

def test_print_table(self):

1287

self.assert_command_from_args([], PrintTableCmd,

1288

verbose=False)

1289

1290

def test_print_table_verbose(self):

1291

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

1292

verbose=True)

1293

1294

def test_print_table_verbose_short(self):

1295

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

1296

verbose=True)

1297

1298

1299

class TestCmd(unittest.TestCase):

1300

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

1301

1302

def setUp(self):

1303

testcase = self

1304

class MockClient(object):

1305

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

1306

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

1307

self.attributes = attributes

1308

self.attributes["Name"] = name

1309

self.calls = []

1310

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

1311

testcase.assertEqual(interface, client_dbus_interface)

1312

testcase.assertEqual(dbus_interface,

1313

dbus.PROPERTIES_IFACE)

1314

self.attributes[propname] = value

1315

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

1316

testcase.assertEqual(interface, client_dbus_interface)

1317

testcase.assertEqual(dbus_interface,

1318

dbus.PROPERTIES_IFACE)

1319

return self.attributes[propname]

1320

def Approve(self, approve, dbus_interface):

1321

testcase.assertEqual(dbus_interface,

1322

client_dbus_interface)

1323

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

1324

dbus_interface)))

1325

self.client = MockClient(

1326

"foo",

1327

KeyID=("92ed150794387c03ce684574b1139a65"

1328

"94a34f895daaaf09fd8ea90a27cddb12"),

1329

Secret=b"secret",

1330

Host="foo.example.org",

1331

Enabled=dbus.Boolean(True),

1332

Timeout=300000,

1333

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

1334

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

1335

Interval=120000,

1336

Fingerprint=("778827225BA7DE539C5A"

1337

"7CFA59CFF7CDBD9A5920"),

1338

CheckerRunning=dbus.Boolean(False),

1339

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

1340

ApprovalPending=dbus.Boolean(False),

1341

ApprovedByDefault=dbus.Boolean(True),

1342

LastApprovalRequest="",

1343

ApprovalDelay=0,

1344

ApprovalDuration=1000,

1345

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

1346

ExtendedTimeout=900000,

1347

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

1348

LastCheckerStatus=0)

1349

self.other_client = MockClient(

1350

"barbar",

1351

KeyID=("0558568eedd67d622f5c83b35a115f79"

1352

"6ab612cff5ad227247e46c2b020f441c"),

1353

Secret=b"secretbar",

1354

Host="192.0.2.3",

1355

Enabled=dbus.Boolean(True),

1356

Timeout=300000,

1357

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

1358

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

1359

Interval=120000,

1360

Fingerprint=("3E393AEAEFB84C7E89E2"

1361

"F547B3A107558FCA3A27"),

1362

CheckerRunning=dbus.Boolean(True),

1363

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

1364

ApprovalPending=dbus.Boolean(False),

1365

ApprovedByDefault=dbus.Boolean(False),

1366

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

1367

ApprovalDelay=30000,

1368

ApprovalDuration=93785000,

1369

Checker=":",

1370

ExtendedTimeout=900000,

1371

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

1372

LastCheckerStatus=-2)

1373

self.clients = collections.OrderedDict(

1374

[

1375

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

1376

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

1377

])

1378

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

1379

1380

@property

1381

def bus(self):

1382

class Bus(object):

1383

@staticmethod

1384

def get_object(client_bus_name, path):

1385

self.assertEqual(client_bus_name, dbus_busname)

1386

return {

1387

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

1388

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

1389

"/clients/foo": self.client,

1390

"/clients/barbar": self.other_client,

1391

}[path]

1392

return Bus()

1393

1394

1395

class TestIsEnabledCmd(TestCmd):

1396

def test_is_enabled(self):

1397

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

1398

properties)

1399

for client, properties

1400

in self.clients.items()))

1401

1402

def test_is_enabled_run_exits_successfully(self):

1403

with self.assertRaises(SystemExit) as e:

1404

IsEnabledCmd().run(self.one_client)

1405

if e.exception.code is not None:

1406

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

1407

else:

1408

self.assertIsNone(e.exception.code)

1409

1410

def test_is_enabled_run_exits_with_failure(self):

1411

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

1412

with self.assertRaises(SystemExit) as e:

1413

IsEnabledCmd().run(self.one_client)

1414

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

1415

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

1416

else:

1417

self.assertIsNotNone(e.exception.code)

1418

1419

1420

class TestApproveCmd(TestCmd):

1421

def test_approve(self):

1422

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

1423

for clientpath in self.clients:

1424

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

1425

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

1426

client.calls)

1427

1428

1429

class TestDenyCmd(TestCmd):

1430

def test_deny(self):

1431

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

1432

for clientpath in self.clients:

1433

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

1434

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

1435

client.calls)

1436

1437

1438

class TestRemoveCmd(TestCmd):

1439

def test_remove(self):

1440

class MockMandos(object):

1441

def __init__(self):

1442

self.calls = []

1443

def RemoveClient(self, dbus_path):

1444

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

1445

mandos = MockMandos()

1446

super(TestRemoveCmd, self).setUp()

1447

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

1448

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

1449

for clientpath in self.clients:

1450

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

1451

mandos.calls)

1452

1453

1454

class TestDumpJSONCmd(TestCmd):

1455

def setUp(self):

1456

self.expected_json = {

1457

"foo": {

1458

"Name": "foo",

1459

"KeyID": ("92ed150794387c03ce684574b1139a65"

1460

"94a34f895daaaf09fd8ea90a27cddb12"),

1461

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

1462

"Enabled": True,

1463

"Timeout": 300000,

1464

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

1465

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

1466

"Interval": 120000,

1467

"Fingerprint": ("778827225BA7DE539C5A"

1468

"7CFA59CFF7CDBD9A5920"),

1469

"CheckerRunning": False,

1470

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

1471

"ApprovalPending": False,

1472

"ApprovedByDefault": True,

1473

"LastApprovalRequest": "",

1474

"ApprovalDelay": 0,

1475

"ApprovalDuration": 1000,

1476

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

1477

"ExtendedTimeout": 900000,

1478

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

1479

"LastCheckerStatus": 0,

1480

},

1481

"barbar": {

1482

"Name": "barbar",

1483

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1484

"6ab612cff5ad227247e46c2b020f441c"),

1485

"Host": "192.0.2.3",

1486

"Enabled": True,

1487

"Timeout": 300000,

1488

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

1489

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

1490

"Interval": 120000,

1491

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1492

"F547B3A107558FCA3A27"),

1493

"CheckerRunning": True,

1494

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

1495

"ApprovalPending": False,

1496

"ApprovedByDefault": False,

1497

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

1498

"ApprovalDelay": 30000,

1499

"ApprovalDuration": 93785000,

1500

"Checker": ":",

1501

"ExtendedTimeout": 900000,

1502

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

1503

"LastCheckerStatus": -2,

1504

},

1505

}

1506

return super(TestDumpJSONCmd, self).setUp()

1507

1508

def test_normal(self):

1509

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

1510

json_data = json.loads(output)

1511

self.assertDictEqual(json_data, self.expected_json)

1512

1513

def test_one_client(self):

1514

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

1515

json_data = json.loads(output)

1516

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

1517

self.assertDictEqual(json_data, expected_json)

1518

1519

1520

class TestPrintTableCmd(TestCmd):

1521

def test_normal(self):

1522

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

1523

expected_output = "\n".join((

1524

"Name Enabled Timeout Last Successful Check",

1525

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

1526

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

1527

))

1528

self.assertEqual(output, expected_output)

1529

1530

def test_verbose(self):

1531

output = PrintTableCmd(verbose=True).output(

1532

self.clients.values())

1533

columns = (

1534

(

1535

"Name ",

1536

"foo ",

1537

"barbar ",

1538

),(

1539

"Enabled ",

1540

"Yes ",

1541

"Yes ",

1542

),(

1543

"Timeout ",

1544

"00:05:00 ",

1545

"00:05:00 ",

1546

),(

1547

"Last Successful Check ",

1548

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

1549

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

1550

),(

1551

"Created ",

1552

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

1553

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

1554

),(

1555

"Interval ",

1556

"00:02:00 ",

1557

"00:02:00 ",

1558

),(

1559

"Host ",

1560

"foo.example.org ",

1561

"192.0.2.3 ",

1562

),(

1563

("Key ID "

1564

" "),

1565

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1566

"ea90a27cddb12 "),

1567

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1568

"46c2b020f441c "),

1569

),(

1570

"Fingerprint ",

1571

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1572

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1573

),(

1574

"Check Is Running ",

1575

"No ",

1576

"Yes ",

1577

),(

1578

"Last Enabled ",

1579

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

1580

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

1581

),(

1582

"Approval Is Pending ",

1583

"No ",

1584

"No ",

1585

),(

1586

"Approved By Default ",

1587

"Yes ",

1588

"No ",

1589

),(

1590

"Last Approval Request ",

1591

" ",

1592

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

1593

),(

1594

"Approval Delay ",

1595

"00:00:00 ",

1596

"00:00:30 ",

1597

),(

1598

"Approval Duration ",

1599

"00:00:01 ",

1600

"1T02:03:05 ",

1601

),(

1602

"Checker ",

1603

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

1604

": ",

1605

),(

1606

"Extended Timeout ",

1607

"00:15:00 ",

1608

"00:15:00 ",

1609

),(

1610

"Expires ",

1611

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

1612

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

1613

),(

1614

"Last Checker Status",

1615

"0 ",

1616

"-2 ",

1617

)

1618

)

1619

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

1620

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

1621

for rows in columns)

1622

for line in range(num_lines))

1623

self.assertEqual(output, expected_output)

1624

1625

def test_one_client(self):

1626

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

1627

expected_output = "\n".join((

1628

"Name Enabled Timeout Last Successful Check",

1629

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

1630

))

1631

self.assertEqual(output, expected_output)

1632

1633

1634

class TestPropertyCmd(TestCmd):

1635

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

1636

def runTest(self):

1637

if not hasattr(self, "command"):

1638

return

1639

values_to_get = getattr(self, "values_to_get",

1640

self.values_to_set)

1641

for value_to_set, value_to_get in zip(self.values_to_set,

1642

values_to_get):

1643

for clientpath in self.clients:

1644

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

1645

old_value = client.attributes[self.propname]

1646

self.assertNotIsInstance(old_value, self.Unique)

1647

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

1648

self.run_command(value_to_set, self.clients)

1649

for clientpath in self.clients:

1650

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

1651

value = client.attributes[self.propname]

1652

self.assertNotIsInstance(value, self.Unique)

1653

self.assertEqual(value, value_to_get)

1654

1655

class Unique(object):

1656

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

1657

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

1658

1659

def run_command(self, value, clients):

1660

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

1661

1662

1663

class TestEnableCmd(TestPropertyCmd):

1664

command = EnableCmd

1665

propname = "Enabled"

1666

values_to_set = [dbus.Boolean(True)]

1667

1668

1669

class TestDisableCmd(TestPropertyCmd):

1670

command = DisableCmd

1671

propname = "Enabled"

1672

values_to_set = [dbus.Boolean(False)]

1673

1674

1675

class TestBumpTimeoutCmd(TestPropertyCmd):

1676

command = BumpTimeoutCmd

1677

propname = "LastCheckedOK"

1678

values_to_set = [""]

1679

1680

1681

class TestStartCheckerCmd(TestPropertyCmd):

1682

command = StartCheckerCmd

1683

propname = "CheckerRunning"

1684

values_to_set = [dbus.Boolean(True)]

1685

1686

1687

class TestStopCheckerCmd(TestPropertyCmd):

1688

command = StopCheckerCmd

1689

propname = "CheckerRunning"

1690

values_to_set = [dbus.Boolean(False)]

1691

1692

1693

class TestApproveByDefaultCmd(TestPropertyCmd):

1694

command = ApproveByDefaultCmd

1695

propname = "ApprovedByDefault"

1696

values_to_set = [dbus.Boolean(True)]

1697

1698

1699

class TestDenyByDefaultCmd(TestPropertyCmd):

1700

command = DenyByDefaultCmd

1701

propname = "ApprovedByDefault"

1702

values_to_set = [dbus.Boolean(False)]

1703

1704

1705

class TestPropertyValueCmd(TestPropertyCmd):

1706

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

1707

1708

def runTest(self):

1709

if type(self) is TestPropertyValueCmd:

1710

return

1711

return super(TestPropertyValueCmd, self).runTest()

1712

1713

def run_command(self, value, clients):

1714

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

1715

1716

1717

class TestSetCheckerCmd(TestPropertyValueCmd):

1718

command = SetCheckerCmd

1719

propname = "Checker"

1720

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

1721

1722

1723

class TestSetHostCmd(TestPropertyValueCmd):

1724

command = SetHostCmd

1725

propname = "Host"

1726

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

1727

1728

1729

class TestSetSecretCmd(TestPropertyValueCmd):

1730

command = SetSecretCmd

1731

propname = "Secret"

1732

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

1733

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

1734

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

1735

1736

1737

class TestSetTimeoutCmd(TestPropertyValueCmd):

1738

command = SetTimeoutCmd

1739

propname = "Timeout"

1740

values_to_set = [datetime.timedelta(),

1741

datetime.timedelta(minutes=5),

1742

datetime.timedelta(seconds=1),

1743

datetime.timedelta(weeks=1),

1744

datetime.timedelta(weeks=52)]

1745

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

1746

1747

1748

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1749

command = SetExtendedTimeoutCmd

1750

propname = "ExtendedTimeout"

1751

values_to_set = [datetime.timedelta(),

1752

datetime.timedelta(minutes=5),

1753

datetime.timedelta(seconds=1),

1754

datetime.timedelta(weeks=1),

1755

datetime.timedelta(weeks=52)]

1756

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

1757

1758

1759

class TestSetIntervalCmd(TestPropertyValueCmd):

1760

command = SetIntervalCmd

1761

propname = "Interval"

1762

values_to_set = [datetime.timedelta(),

1763

datetime.timedelta(minutes=5),

1764

datetime.timedelta(seconds=1),

1765

datetime.timedelta(weeks=1),

1766

datetime.timedelta(weeks=52)]

1767

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

1768

1769

1770

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1771

command = SetApprovalDelayCmd

1772

propname = "ApprovalDelay"

1773

values_to_set = [datetime.timedelta(),

1774

datetime.timedelta(minutes=5),

1775

datetime.timedelta(seconds=1),

1776

datetime.timedelta(weeks=1),

1777

datetime.timedelta(weeks=52)]

1778

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

1779

1780

1781

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1782

command = SetApprovalDurationCmd

1783

propname = "ApprovalDuration"

1784

values_to_set = [datetime.timedelta(),

1785

datetime.timedelta(minutes=5),

1786

datetime.timedelta(seconds=1),

1787

datetime.timedelta(weeks=1),

1788

datetime.timedelta(weeks=52)]

1789

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

1790

1791

1792

1793

def should_only_run_tests():

1794

parser = argparse.ArgumentParser(add_help=False)

1795

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

1796

args, unknown_args = parser.parse_known_args()

1797

run_tests = args.check

1798

if run_tests:

1799

# Remove --check argument from sys.argv

1800

sys.argv[1:] = unknown_args

1801

return run_tests

1802

1803

# Add all tests from doctest strings

1804

def load_tests(loader, tests, none):

1805

import doctest

1806

tests.addTests(doctest.DocTestSuite())

1807

return tests

344

1808

345

1809

if __name__ == "__main__":

346

main()

1810

try:

1811

if should_only_run_tests():

1812

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

1813

unittest.main()

1814

else:

1815

main()

1816

finally:

1817

logging.shutdown()