/mandos/trunk : revision 1075

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

from future_builtins import *

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

30

34

31

35

import sys

32

import dbus

33

36

import argparse

34

37

import locale

35

38

import datetime

36

39

import re

37

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

38

64

39

65

locale.setlocale(locale.LC_ALL, "")

40

66

41

tablewords = {

42

"Name": "Name",

43

"Enabled": "Enabled",

44

"Timeout": "Timeout",

45

"LastCheckedOK": "Last Successful Check",

46

"LastApprovalRequest": "Last Approval Request",

47

"Created": "Created",

48

"Interval": "Interval",

49

"Host": "Host",

50

"Fingerprint": "Fingerprint",

51

"CheckerRunning": "Check Is Running",

52

"LastEnabled": "Last Enabled",

53

"ApprovalPending": "Approval Is Pending",

54

"ApprovedByDefault": "Approved By Default",

55

"ApprovalDelay": "Approval Delay",

56

"ApprovalDuration": "Approval Duration",

57

"Checker": "Checker",

58

"ExtendedTimeout" : "Extended Timeout"

59

}

60

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

61

domain = "se.recompile"

62

busname = domain + ".Mandos"

63

server_path = "/"

64

server_interface = domain + ".Mandos"

65

client_interface = domain + ".Mandos.Client"

66

version = "1.5.4"

67

68

def timedelta_to_milliseconds(td):

69

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

70

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

71

+ (td.seconds * 1000)

72

+ (td.microseconds // 1000))

73

74

def milliseconds_to_string(ms):

75

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

76

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

77

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

78

hours = td.seconds // 3600,

79

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

80

seconds = td.seconds % 60,

81

))

82

83

def string_to_delta(interval):

84

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

85

86

>>> string_to_delta("7d")

87

datetime.timedelta(7)

88

>>> string_to_delta("60s")

89

datetime.timedelta(0, 60)

90

>>> string_to_delta("60m")

91

datetime.timedelta(0, 3600)

92

>>> string_to_delta("24h")

93

datetime.timedelta(1)

94

>>> string_to_delta("1w")

95

datetime.timedelta(7)

96

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

97

datetime.timedelta(0, 330)

98

"""

99

value = datetime.timedelta(0)

100

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

101

102

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

103

if suffix == "d":

104

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

105

elif suffix == "s":

106

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

107

elif suffix == "m":

108

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

109

elif suffix == "h":

110

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

111

elif suffix == "w":

112

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

113

elif suffix == "":

114

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

115

return value

116

117

def print_clients(clients, keywords):

118

def valuetostring(value, keyword):

119

if type(value) is dbus.Boolean:

120

return "Yes" if value else "No"

121

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

122

"ApprovalDuration", "ExtendedTimeout"):

123

return milliseconds_to_string(value)

124

return unicode(value)

125

126

# Create format string to print table rows

127

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

128

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

129

max(len(valuetostring(client[key],

130

key))

131

for client in

132

clients)),

133

key = key) for key in keywords)

134

# Print header line

135

print(format_string.format(**tablewords))

136

for client in clients:

137

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

138

valuetostring(client[key],

139

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.extended_timeout is not None,

153

options.interval is not None,

154

options.approved_by_default is not None,

155

options.approval_delay is not None,

156

options.approval_duration is not None,

157

options.host is not None,

158

options.secret is not None,

159

options.approve,

160

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

161

81

162

82

def main():

163

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

164

135

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

165

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

136

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

166

137

help="show version number and exit")

167

138

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

168

139

help="Select all clients")

169

140

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

170

141

help="Print all fields")

171

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

172

help="Enable client")

173

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

174

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

175

150

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

176

151

help="Bump timeout for client")

177

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

178

help="Start checker for client")

179

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

180

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

181

159

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

182

160

help="Check if client is enabled")

183

161

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

184

162

help="Remove client")

185

163

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

186

164

help="Set checker command for client")

187

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

165

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

188

166

help="Set timeout for client")

189

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

167

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

190

168

help="Set extended timeout for client")

191

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

169

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

192

170

help="Set checker interval for client")

193

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

194

default=None, dest="approved_by_default",

195

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

196

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

197

dest="approved_by_default",

198

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

199

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,

200

181

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

201

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

182

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

202

183

help="Set duration of one client approval")

203

184

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

204

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

185

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

186

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

205

187

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

206

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

207

help="Approve any current client request")

208

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

209

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

210

198

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

211

options = parser.parse_args()

212

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

213

409

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

214

410

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

215

411

if options.verbose and has_actions(options):

216

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

217

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

218

416

if options.all and not has_actions(options):

219

417

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

220

221

try:

222

bus = dbus.SystemBus()

223

mandos_dbus_objc = bus.get_object(busname, server_path)

224

except dbus.exceptions.DBusException:

225

print("Could not connect to Mandos server",

226

file=sys.stderr)

227

sys.exit(1)

228

229

mandos_serv = dbus.Interface(mandos_dbus_objc,

230

dbus_interface = server_interface)

231

232

#block stderr since dbus library prints to stderr

233

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

234

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

235

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

236

os.close(null)

237

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)

238

947

try:

239

mandos_clients = mandos_serv.GetAllClientsWithProperties()

948

yield

240

949

finally:

241

#restore stderr

950

# restore stderr

242

951

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

243

952

os.close(stderrcopy)

244

except dbus.exceptions.DBusException:

245

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

246

file=sys.stderr)

247

sys.exit(1)

248

249

# Compile dict of (clients: properties) to process

250

clients={}

251

252

if options.all or not options.client:

253

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

254

for path, properties in

255

mandos_clients.iteritems())

256

else:

257

for name in options.client:

258

for path, client in mandos_clients.iteritems():

259

if client["Name"] == name:

260

client_objc = bus.get_object(busname, path)

261

clients[client_objc] = client

262

break

263

else:

264

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

265

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

266

sys.exit(1)

267

268

if not has_actions(options) and clients:

269

if options.verbose:

270

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

271

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

272

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

273

"LastEnabled", "ApprovalPending",

274

"ApprovedByDefault",

275

"LastApprovalRequest", "ApprovalDelay",

276

"ApprovalDuration", "Checker",

277

"ExtendedTimeout")

278

else:

279

keywords = defaultkeywords

280

281

print_clients(clients.values(), keywords)

282

else:

283

# Process each client in the list by all selected options

284

for client in clients:

285

def set_client_prop(prop, value):

286

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

287

client.Set(client_interface, prop, value,

288

dbus_interface=dbus.PROPERTIES_IFACE)

289

def set_client_prop_ms(prop, value):

290

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

291

from a string to milliseconds."""

292

set_client_prop(prop,

293

timedelta_to_milliseconds

294

(string_to_delta(value)))

295

if options.remove:

296

mandos_serv.RemoveClient(client.__dbus_object_path__)

297

if options.enable:

298

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

299

if options.disable:

300

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

301

if options.bump_timeout:

302

set_client_prop("LastCheckedOK", "")

303

if options.start_checker:

304

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

305

if options.stop_checker:

306

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

307

if options.is_enabled:

308

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

309

"Enabled",

310

dbus_interface=

311

dbus.PROPERTIES_IFACE)

312

else 1)

313

if options.checker is not None:

314

set_client_prop("Checker", options.checker)

315

if options.host is not None:

316

set_client_prop("Host", options.host)

317

if options.interval is not None:

318

set_client_prop_ms("Interval", options.interval)

319

if options.approval_delay is not None:

320

set_client_prop_ms("ApprovalDelay",

321

options.approval_delay)

322

if options.approval_duration is not None:

323

set_client_prop_ms("ApprovalDuration",

324

options.approval_duration)

325

if options.timeout is not None:

326

set_client_prop_ms("Timeout", options.timeout)

327

if options.extended_timeout is not None:

328

set_client_prop_ms("ExtendedTimeout",

329

options.extended_timeout)

330

if options.secret is not None:

331

set_client_prop("Secret",

332

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

333

if options.approved_by_default is not None:

334

set_client_prop("ApprovedByDefault",

335

dbus.Boolean(options

336

.approved_by_default))

337

if options.approve:

338

client.Approve(dbus.Boolean(True),

339

dbus_interface=client_interface)

340

elif options.deny:

341

client.Approve(dbus.Boolean(False),

342

dbus_interface=client_interface)

953

954

def check_option_syntax(self, options):

955

check_option_syntax(self.parser, options)

956

957

def test_actions_all_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.all = True

962

options.verbose = True

963

with self.assertParseError():

964

self.check_option_syntax(options)

965

966

def test_actions_with_client_conflicts_with_verbose(self):

967

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

968

options = self.parser.parse_args()

969

setattr(options, action, value)

970

options.verbose = True

971

options.client = ["foo"]

972

with self.assertParseError():

973

self.check_option_syntax(options)

974

975

def test_dump_json_conflicts_with_verbose(self):

976

options = self.parser.parse_args()

977

options.dump_json = True

978

options.verbose = True

979

with self.assertParseError():

980

self.check_option_syntax(options)

981

982

def test_dump_json_conflicts_with_action(self):

983

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

984

options = self.parser.parse_args()

985

setattr(options, action, value)

986

options.dump_json = True

987

with self.assertParseError():

988

self.check_option_syntax(options)

989

990

def test_all_can_not_be_alone(self):

991

options = self.parser.parse_args()

992

options.all = True

993

with self.assertParseError():

994

self.check_option_syntax(options)

995

996

def test_all_is_ok_with_any_action(self):

997

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

998

options = self.parser.parse_args()

999

setattr(options, action, value)

1000

options.all = True

1001

self.check_option_syntax(options)

1002

1003

def test_any_action_is_ok_with_one_client(self):

1004

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

1005

options = self.parser.parse_args()

1006

setattr(options, action, value)

1007

options.client = ["foo"]

1008

self.check_option_syntax(options)

1009

1010

def test_actions_except_is_enabled_are_ok_with_two_clients(self):

1011

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

1012

if action == "is_enabled":

1013

continue

1014

options = self.parser.parse_args()

1015

setattr(options, action, value)

1016

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

1017

self.check_option_syntax(options)

1018

1019

def test_is_enabled_fails_without_client(self):

1020

options = self.parser.parse_args()

1021

options.is_enabled = True

1022

with self.assertParseError():

1023

self.check_option_syntax(options)

1024

1025

def test_is_enabled_fails_with_two_clients(self):

1026

options = self.parser.parse_args()

1027

options.is_enabled = True

1028

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

1029

with self.assertParseError():

1030

self.check_option_syntax(options)

1031

1032

def test_remove_can_only_be_combined_with_action_deny(self):

1033

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

1034

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

1035

continue

1036

options = self.parser.parse_args()

1037

setattr(options, action, value)

1038

options.all = True

1039

options.remove = True

1040

with self.assertParseError():

1041

self.check_option_syntax(options)

1042

1043

1044

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1045

def test_calls_and_returns_get_object_on_bus(self):

1046

class MockBus(object):

1047

called = False

1048

def get_object(mockbus_self, busname, dbus_path):

1049

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

1050

# this MockBus instance is in "mockbus_self".

1051

self.assertEqual(busname, dbus_busname)

1052

self.assertEqual(dbus_path, server_dbus_path)

1053

mockbus_self.called = True

1054

return mockbus_self

1055

1056

mockbus = get_mandos_dbus_object(bus=MockBus())

1057

self.assertIsInstance(mockbus, MockBus)

1058

self.assertTrue(mockbus.called)

1059

1060

def test_logs_and_exits_on_dbus_error(self):

1061

class MockBusFailing(object):

1062

def get_object(self, busname, dbus_path):

1063

raise dbus.exceptions.DBusException("Test")

1064

1065

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

1066

with self.assertRaises(SystemExit) as e:

1067

bus = get_mandos_dbus_object(bus=MockBusFailing())

1068

1069

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

1070

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

1071

else:

1072

self.assertIsNotNone(e.exception.code)

1073

1074

1075

class Test_get_managed_objects(TestCaseWithAssertLogs):

1076

def test_calls_and_returns_GetManagedObjects(self):

1077

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

1078

class MockObjectManager(object):

1079

def GetManagedObjects(self):

1080

return managed_objects

1081

retval = get_managed_objects(MockObjectManager())

1082

self.assertDictEqual(managed_objects, retval)

1083

1084

def test_logs_and_exits_on_dbus_error(self):

1085

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

1086

1087

class MockObjectManagerFailing(object):

1088

def GetManagedObjects(self):

1089

dbus_logger.error("Test")

1090

raise dbus.exceptions.DBusException("Test")

1091

1092

class CountingHandler(logging.Handler):

1093

count = 0

1094

def emit(self, record):

1095

self.count += 1

1096

1097

counting_handler = CountingHandler()

1098

1099

dbus_logger.addHandler(counting_handler)

1100

1101

try:

1102

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

1103

with self.assertRaises(SystemExit) as e:

1104

get_managed_objects(MockObjectManagerFailing())

1105

finally:

1106

dbus_logger.removeFilter(counting_handler)

1107

1108

# Make sure the dbus logger was suppressed

1109

self.assertEqual(counting_handler.count, 0)

1110

1111

# Test that the dbus_logger still works

1112

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

1113

dbus_logger.error("Test")

1114

1115

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

1116

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

1117

else:

1118

self.assertIsNotNone(e.exception.code)

1119

1120

1121

class Test_commands_from_options(unittest.TestCase):

1122

def setUp(self):

1123

self.parser = argparse.ArgumentParser()

1124

add_command_line_options(self.parser)

1125

1126

def test_is_enabled(self):

1127

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

1128

IsEnabledCmd)

1129

1130

def assert_command_from_args(self, args, command_cls,

1131

**cmd_attrs):

1132

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

1133

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

1134

options = self.parser.parse_args(args)

1135

check_option_syntax(self.parser, options)

1136

commands = commands_from_options(options)

1137

self.assertEqual(len(commands), 1)

1138

command = commands[0]

1139

self.assertIsInstance(command, command_cls)

1140

for key, value in cmd_attrs.items():

1141

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

1142

1143

def test_is_enabled_short(self):

1144

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

1145

1146

def test_approve(self):

1147

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

1148

ApproveCmd)

1149

1150

def test_approve_short(self):

1151

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

1152

1153

def test_deny(self):

1154

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

1155

1156

def test_deny_short(self):

1157

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

1158

1159

def test_remove(self):

1160

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

1161

RemoveCmd)

1162

1163

def test_deny_before_remove(self):

1164

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

1165

"foo"])

1166

check_option_syntax(self.parser, options)

1167

commands = commands_from_options(options)

1168

self.assertEqual(len(commands), 2)

1169

self.assertIsInstance(commands[0], DenyCmd)

1170

self.assertIsInstance(commands[1], RemoveCmd)

1171

1172

def test_deny_before_remove_reversed(self):

1173

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

1174

"--all"])

1175

check_option_syntax(self.parser, options)

1176

commands = commands_from_options(options)

1177

self.assertEqual(len(commands), 2)

1178

self.assertIsInstance(commands[0], DenyCmd)

1179

self.assertIsInstance(commands[1], RemoveCmd)

1180

1181

def test_remove_short(self):

1182

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

1183

1184

def test_dump_json(self):

1185

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

1186

1187

def test_enable(self):

1188

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

1189

1190

def test_enable_short(self):

1191

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

1192

1193

def test_disable(self):

1194

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

1195

DisableCmd)

1196

1197

def test_disable_short(self):

1198

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

1199

1200

def test_bump_timeout(self):

1201

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

1202

BumpTimeoutCmd)

1203

1204

def test_bump_timeout_short(self):

1205

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

1206

1207

def test_start_checker(self):

1208

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

1209

StartCheckerCmd)

1210

1211

def test_stop_checker(self):

1212

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

1213

StopCheckerCmd)

1214

1215

def test_approve_by_default(self):

1216

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

1217

ApproveByDefaultCmd)

1218

1219

def test_deny_by_default(self):

1220

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

1221

DenyByDefaultCmd)

1222

1223

def test_checker(self):

1224

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

1225

SetCheckerCmd, value_to_set=":")

1226

1227

def test_checker_empty(self):

1228

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

1229

SetCheckerCmd, value_to_set="")

1230

1231

def test_checker_short(self):

1232

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

1233

SetCheckerCmd, value_to_set=":")

1234

1235

def test_host(self):

1236

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

1237

"foo"], SetHostCmd,

1238

value_to_set="foo.example.org")

1239

1240

def test_host_short(self):

1241

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

1242

"foo"], SetHostCmd,

1243

value_to_set="foo.example.org")

1244

1245

def test_secret_devnull(self):

1246

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

1247

"foo"], SetSecretCmd,

1248

value_to_set=b"")

1249

1250

def test_secret_tempfile(self):

1251

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

1252

value = b"secret\0xyzzy\nbar"

1253

f.write(value)

1254

f.seek(0)

1255

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

1256

"foo"], SetSecretCmd,

1257

value_to_set=value)

1258

1259

def test_secret_devnull_short(self):

1260

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

1261

SetSecretCmd, value_to_set=b"")

1262

1263

def test_secret_tempfile_short(self):

1264

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

1265

value = b"secret\0xyzzy\nbar"

1266

f.write(value)

1267

f.seek(0)

1268

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

1269

SetSecretCmd,

1270

value_to_set=value)

1271

1272

def test_timeout(self):

1273

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

1274

SetTimeoutCmd,

1275

value_to_set=300000)

1276

1277

def test_timeout_short(self):

1278

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

1279

SetTimeoutCmd,

1280

value_to_set=300000)

1281

1282

def test_extended_timeout(self):

1283

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

1284

"foo"],

1285

SetExtendedTimeoutCmd,

1286

value_to_set=900000)

1287

1288

def test_interval(self):

1289

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

1290

SetIntervalCmd,

1291

value_to_set=120000)

1292

1293

def test_interval_short(self):

1294

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

1295

SetIntervalCmd,

1296

value_to_set=120000)

1297

1298

def test_approval_delay(self):

1299

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

1300

"foo"], SetApprovalDelayCmd,

1301

value_to_set=30000)

1302

1303

def test_approval_duration(self):

1304

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

1305

"foo"], SetApprovalDurationCmd,

1306

value_to_set=1000)

1307

1308

def test_print_table(self):

1309

self.assert_command_from_args([], PrintTableCmd,

1310

verbose=False)

1311

1312

def test_print_table_verbose(self):

1313

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

1314

verbose=True)

1315

1316

def test_print_table_verbose_short(self):

1317

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

1318

verbose=True)

1319

1320

1321

class TestCmd(unittest.TestCase):

1322

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

1323

1324

def setUp(self):

1325

testcase = self

1326

class MockClient(object):

1327

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

1328

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

1329

self.attributes = attributes

1330

self.attributes["Name"] = name

1331

self.calls = []

1332

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

1333

testcase.assertEqual(interface, client_dbus_interface)

1334

testcase.assertEqual(dbus_interface,

1335

dbus.PROPERTIES_IFACE)

1336

self.attributes[propname] = value

1337

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

1338

testcase.assertEqual(interface, client_dbus_interface)

1339

testcase.assertEqual(dbus_interface,

1340

dbus.PROPERTIES_IFACE)

1341

return self.attributes[propname]

1342

def Approve(self, approve, dbus_interface):

1343

testcase.assertEqual(dbus_interface,

1344

client_dbus_interface)

1345

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

1346

dbus_interface)))

1347

self.client = MockClient(

1348

"foo",

1349

KeyID=("92ed150794387c03ce684574b1139a65"

1350

"94a34f895daaaf09fd8ea90a27cddb12"),

1351

Secret=b"secret",

1352

Host="foo.example.org",

1353

Enabled=dbus.Boolean(True),

1354

Timeout=300000,

1355

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

1356

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

1357

Interval=120000,

1358

Fingerprint=("778827225BA7DE539C5A"

1359

"7CFA59CFF7CDBD9A5920"),

1360

CheckerRunning=dbus.Boolean(False),

1361

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

1362

ApprovalPending=dbus.Boolean(False),

1363

ApprovedByDefault=dbus.Boolean(True),

1364

LastApprovalRequest="",

1365

ApprovalDelay=0,

1366

ApprovalDuration=1000,

1367

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

1368

ExtendedTimeout=900000,

1369

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

1370

LastCheckerStatus=0)

1371

self.other_client = MockClient(

1372

"barbar",

1373

KeyID=("0558568eedd67d622f5c83b35a115f79"

1374

"6ab612cff5ad227247e46c2b020f441c"),

1375

Secret=b"secretbar",

1376

Host="192.0.2.3",

1377

Enabled=dbus.Boolean(True),

1378

Timeout=300000,

1379

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

1380

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

1381

Interval=120000,

1382

Fingerprint=("3E393AEAEFB84C7E89E2"

1383

"F547B3A107558FCA3A27"),

1384

CheckerRunning=dbus.Boolean(True),

1385

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

1386

ApprovalPending=dbus.Boolean(False),

1387

ApprovedByDefault=dbus.Boolean(False),

1388

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

1389

ApprovalDelay=30000,

1390

ApprovalDuration=93785000,

1391

Checker=":",

1392

ExtendedTimeout=900000,

1393

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

1394

LastCheckerStatus=-2)

1395

self.clients = collections.OrderedDict(

1396

[

1397

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

1398

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

1399

])

1400

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

1401

1402

@property

1403

def bus(self):

1404

class Bus(object):

1405

@staticmethod

1406

def get_object(client_bus_name, path):

1407

self.assertEqual(client_bus_name, dbus_busname)

1408

return {

1409

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

1410

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

1411

"/clients/foo": self.client,

1412

"/clients/barbar": self.other_client,

1413

}[path]

1414

return Bus()

1415

1416

1417

class TestIsEnabledCmd(TestCmd):

1418

def test_is_enabled(self):

1419

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

1420

properties)

1421

for client, properties

1422

in self.clients.items()))

1423

1424

def test_is_enabled_run_exits_successfully(self):

1425

with self.assertRaises(SystemExit) as e:

1426

IsEnabledCmd().run(self.one_client)

1427

if e.exception.code is not None:

1428

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

1429

else:

1430

self.assertIsNone(e.exception.code)

1431

1432

def test_is_enabled_run_exits_with_failure(self):

1433

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

1434

with self.assertRaises(SystemExit) as e:

1435

IsEnabledCmd().run(self.one_client)

1436

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

1437

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

1438

else:

1439

self.assertIsNotNone(e.exception.code)

1440

1441

1442

class TestApproveCmd(TestCmd):

1443

def test_approve(self):

1444

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

1445

for clientpath in self.clients:

1446

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

1447

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

1448

client.calls)

1449

1450

1451

class TestDenyCmd(TestCmd):

1452

def test_deny(self):

1453

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

1454

for clientpath in self.clients:

1455

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

1456

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

1457

client.calls)

1458

1459

1460

class TestRemoveCmd(TestCmd):

1461

def test_remove(self):

1462

class MockMandos(object):

1463

def __init__(self):

1464

self.calls = []

1465

def RemoveClient(self, dbus_path):

1466

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

1467

mandos = MockMandos()

1468

super(TestRemoveCmd, self).setUp()

1469

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

1470

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

1471

for clientpath in self.clients:

1472

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

1473

mandos.calls)

1474

1475

1476

class TestDumpJSONCmd(TestCmd):

1477

def setUp(self):

1478

self.expected_json = {

1479

"foo": {

1480

"Name": "foo",

1481

"KeyID": ("92ed150794387c03ce684574b1139a65"

1482

"94a34f895daaaf09fd8ea90a27cddb12"),

1483

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

1484

"Enabled": True,

1485

"Timeout": 300000,

1486

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

1487

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

1488

"Interval": 120000,

1489

"Fingerprint": ("778827225BA7DE539C5A"

1490

"7CFA59CFF7CDBD9A5920"),

1491

"CheckerRunning": False,

1492

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

1493

"ApprovalPending": False,

1494

"ApprovedByDefault": True,

1495

"LastApprovalRequest": "",

1496

"ApprovalDelay": 0,

1497

"ApprovalDuration": 1000,

1498

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

1499

"ExtendedTimeout": 900000,

1500

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

1501

"LastCheckerStatus": 0,

1502

},

1503

"barbar": {

1504

"Name": "barbar",

1505

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1506

"6ab612cff5ad227247e46c2b020f441c"),

1507

"Host": "192.0.2.3",

1508

"Enabled": True,

1509

"Timeout": 300000,

1510

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

1511

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

1512

"Interval": 120000,

1513

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1514

"F547B3A107558FCA3A27"),

1515

"CheckerRunning": True,

1516

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

1517

"ApprovalPending": False,

1518

"ApprovedByDefault": False,

1519

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

1520

"ApprovalDelay": 30000,

1521

"ApprovalDuration": 93785000,

1522

"Checker": ":",

1523

"ExtendedTimeout": 900000,

1524

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

1525

"LastCheckerStatus": -2,

1526

},

1527

}

1528

return super(TestDumpJSONCmd, self).setUp()

1529

1530

def test_normal(self):

1531

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

1532

json_data = json.loads(output)

1533

self.assertDictEqual(json_data, self.expected_json)

1534

1535

def test_one_client(self):

1536

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

1537

json_data = json.loads(output)

1538

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

1539

self.assertDictEqual(json_data, expected_json)

1540

1541

1542

class TestPrintTableCmd(TestCmd):

1543

def test_normal(self):

1544

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

1545

expected_output = "\n".join((

1546

"Name Enabled Timeout Last Successful Check",

1547

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

1548

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

1549

))

1550

self.assertEqual(output, expected_output)

1551

1552

def test_verbose(self):

1553

output = PrintTableCmd(verbose=True).output(

1554

self.clients.values())

1555

columns = (

1556

(

1557

"Name ",

1558

"foo ",

1559

"barbar ",

1560

),(

1561

"Enabled ",

1562

"Yes ",

1563

"Yes ",

1564

),(

1565

"Timeout ",

1566

"00:05:00 ",

1567

"00:05:00 ",

1568

),(

1569

"Last Successful Check ",

1570

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

1571

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

1572

),(

1573

"Created ",

1574

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

1575

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

1576

),(

1577

"Interval ",

1578

"00:02:00 ",

1579

"00:02:00 ",

1580

),(

1581

"Host ",

1582

"foo.example.org ",

1583

"192.0.2.3 ",

1584

),(

1585

("Key ID "

1586

" "),

1587

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1588

"ea90a27cddb12 "),

1589

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1590

"46c2b020f441c "),

1591

),(

1592

"Fingerprint ",

1593

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1594

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1595

),(

1596

"Check Is Running ",

1597

"No ",

1598

"Yes ",

1599

),(

1600

"Last Enabled ",

1601

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

1602

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

1603

),(

1604

"Approval Is Pending ",

1605

"No ",

1606

"No ",

1607

),(

1608

"Approved By Default ",

1609

"Yes ",

1610

"No ",

1611

),(

1612

"Last Approval Request ",

1613

" ",

1614

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

1615

),(

1616

"Approval Delay ",

1617

"00:00:00 ",

1618

"00:00:30 ",

1619

),(

1620

"Approval Duration ",

1621

"00:00:01 ",

1622

"1T02:03:05 ",

1623

),(

1624

"Checker ",

1625

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

1626

": ",

1627

),(

1628

"Extended Timeout ",

1629

"00:15:00 ",

1630

"00:15:00 ",

1631

),(

1632

"Expires ",

1633

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

1634

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

1635

),(

1636

"Last Checker Status",

1637

"0 ",

1638

"-2 ",

1639

)

1640

)

1641

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

1642

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

1643

for rows in columns)

1644

for line in range(num_lines))

1645

self.assertEqual(output, expected_output)

1646

1647

def test_one_client(self):

1648

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

1649

expected_output = "\n".join((

1650

"Name Enabled Timeout Last Successful Check",

1651

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

1652

))

1653

self.assertEqual(output, expected_output)

1654

1655

1656

class TestPropertyCmd(TestCmd):

1657

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

1658

def runTest(self):

1659

if not hasattr(self, "command"):

1660

return

1661

values_to_get = getattr(self, "values_to_get",

1662

self.values_to_set)

1663

for value_to_set, value_to_get in zip(self.values_to_set,

1664

values_to_get):

1665

for clientpath in self.clients:

1666

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

1667

old_value = client.attributes[self.propname]

1668

self.assertNotIsInstance(old_value, self.Unique)

1669

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

1670

self.run_command(value_to_set, self.clients)

1671

for clientpath in self.clients:

1672

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

1673

value = client.attributes[self.propname]

1674

self.assertNotIsInstance(value, self.Unique)

1675

self.assertEqual(value, value_to_get)

1676

1677

class Unique(object):

1678

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

1679

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

1680

1681

def run_command(self, value, clients):

1682

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

1683

1684

1685

class TestEnableCmd(TestPropertyCmd):

1686

command = EnableCmd

1687

propname = "Enabled"

1688

values_to_set = [dbus.Boolean(True)]

1689

1690

1691

class TestDisableCmd(TestPropertyCmd):

1692

command = DisableCmd

1693

propname = "Enabled"

1694

values_to_set = [dbus.Boolean(False)]

1695

1696

1697

class TestBumpTimeoutCmd(TestPropertyCmd):

1698

command = BumpTimeoutCmd

1699

propname = "LastCheckedOK"

1700

values_to_set = [""]

1701

1702

1703

class TestStartCheckerCmd(TestPropertyCmd):

1704

command = StartCheckerCmd

1705

propname = "CheckerRunning"

1706

values_to_set = [dbus.Boolean(True)]

1707

1708

1709

class TestStopCheckerCmd(TestPropertyCmd):

1710

command = StopCheckerCmd

1711

propname = "CheckerRunning"

1712

values_to_set = [dbus.Boolean(False)]

1713

1714

1715

class TestApproveByDefaultCmd(TestPropertyCmd):

1716

command = ApproveByDefaultCmd

1717

propname = "ApprovedByDefault"

1718

values_to_set = [dbus.Boolean(True)]

1719

1720

1721

class TestDenyByDefaultCmd(TestPropertyCmd):

1722

command = DenyByDefaultCmd

1723

propname = "ApprovedByDefault"

1724

values_to_set = [dbus.Boolean(False)]

1725

1726

1727

class TestPropertyValueCmd(TestPropertyCmd):

1728

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

1729

1730

def runTest(self):

1731

if type(self) is TestPropertyValueCmd:

1732

return

1733

return super(TestPropertyValueCmd, self).runTest()

1734

1735

def run_command(self, value, clients):

1736

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

1737

1738

1739

class TestSetCheckerCmd(TestPropertyValueCmd):

1740

command = SetCheckerCmd

1741

propname = "Checker"

1742

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

1743

1744

1745

class TestSetHostCmd(TestPropertyValueCmd):

1746

command = SetHostCmd

1747

propname = "Host"

1748

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

1749

1750

1751

class TestSetSecretCmd(TestPropertyValueCmd):

1752

command = SetSecretCmd

1753

propname = "Secret"

1754

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

1755

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

1756

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

1757

1758

1759

class TestSetTimeoutCmd(TestPropertyValueCmd):

1760

command = SetTimeoutCmd

1761

propname = "Timeout"

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 TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1771

command = SetExtendedTimeoutCmd

1772

propname = "ExtendedTimeout"

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 TestSetIntervalCmd(TestPropertyValueCmd):

1782

command = SetIntervalCmd

1783

propname = "Interval"

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

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1793

command = SetApprovalDelayCmd

1794

propname = "ApprovalDelay"

1795

values_to_set = [datetime.timedelta(),

1796

datetime.timedelta(minutes=5),

1797

datetime.timedelta(seconds=1),

1798

datetime.timedelta(weeks=1),

1799

datetime.timedelta(weeks=52)]

1800

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

1801

1802

1803

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1804

command = SetApprovalDurationCmd

1805

propname = "ApprovalDuration"

1806

values_to_set = [datetime.timedelta(),

1807

datetime.timedelta(minutes=5),

1808

datetime.timedelta(seconds=1),

1809

datetime.timedelta(weeks=1),

1810

datetime.timedelta(weeks=52)]

1811

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

1812

1813

1814

1815

def should_only_run_tests():

1816

parser = argparse.ArgumentParser(add_help=False)

1817

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

1818

args, unknown_args = parser.parse_known_args()

1819

run_tests = args.check

1820

if run_tests:

1821

# Remove --check argument from sys.argv

1822

sys.argv[1:] = unknown_args

1823

return run_tests

1824

1825

# Add all tests from doctest strings

1826

def load_tests(loader, tests, none):

1827

import doctest

1828

tests.addTests(doctest.DocTestSuite())

1829

return tests

343

1830

344

1831

if __name__ == "__main__":

345

main()

1832

try:

1833

if should_only_run_tests():

1834

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

1835

unittest.main()

1836

else:

1837

main()

1838

finally:

1839

logging.shutdown()