/mandos/trunk : revision 1083

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

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

23

#

22

24

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

23

#

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

"ExtendedTimeout" : "Extended Timeout"

56

}

57

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

58

domain = "se.recompile"

59

busname = domain + ".Mandos"

60

server_path = "/"

61

server_interface = domain + ".Mandos"

62

client_interface = domain + ".Mandos.Client"

63

version = "1.4.0"

64

65

def timedelta_to_milliseconds(td):

66

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

67

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

68

+ (td.seconds * 1000)

69

+ (td.microseconds // 1000))

70

71

def milliseconds_to_string(ms):

72

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

73

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

74

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

75

"hours": td.seconds // 3600,

76

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

77

"seconds": td.seconds % 60,

78

})

79

80

def string_to_delta(interval):

81

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

82

83

>>> string_to_delta("7d")

84

datetime.timedelta(7)

85

>>> string_to_delta("60s")

86

datetime.timedelta(0, 60)

87

>>> string_to_delta("60m")

88

datetime.timedelta(0, 3600)

89

>>> string_to_delta("24h")

90

datetime.timedelta(1)

91

>>> string_to_delta("1w")

92

datetime.timedelta(7)

93

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

94

datetime.timedelta(0, 330)

95

"""

96

timevalue = datetime.timedelta(0)

97

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

98

99

for s in regexp.findall(interval):

100

try:

101

suffix = unicode(s[-1])

102

value = int(s[:-1])

103

if suffix == "d":

104

delta = datetime.timedelta(value)

105

elif suffix == "s":

106

delta = datetime.timedelta(0, value)

107

elif suffix == "m":

108

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

109

elif suffix == "h":

110

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

111

elif suffix == "w":

112

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

113

else:

114

raise ValueError

115

except (ValueError, IndexError):

116

raise ValueError

117

timevalue += delta

118

return timevalue

119

120

def print_clients(clients, keywords):

121

def valuetostring(value, keyword):

122

if type(value) is dbus.Boolean:

123

return "Yes" if value else "No"

124

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

125

"ApprovalDuration"):

126

return milliseconds_to_string(value)

127

return unicode(value)

128

129

# Create format string to print table rows

130

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

131

max(len(tablewords[key]),

132

max(len(valuetostring(client[key],

133

key))

134

for client in

135

clients))

136

for key in keywords)

137

# Print header line

138

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

139

for client in clients:

140

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

141

for key in keywords))

142

143

def has_actions(options):

144

return any((options.enable,

145

options.disable,

146

options.bump_timeout,

147

options.start_checker,

148

options.stop_checker,

149

options.is_enabled,

150

options.remove,

151

options.checker is not None,

152

options.timeout is not None,

153

options.extended_timeout is not None,

154

options.interval is not None,

155

options.approved_by_default is not None,

156

options.approval_delay is not None,

157

options.approval_duration is not None,

158

options.host is not None,

159

options.secret is not None,

160

options.approve,

161

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

162

81

163

82

def main():

164

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

commands = commands_from_options(options)

129

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

165

135

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

166

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

136

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

167

137

help="show version number and exit")

168

138

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

169

139

help="Select all clients")

170

140

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

171

141

help="Print all fields")

172

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

173

help="Enable client")

174

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

175

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

176

150

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

177

151

help="Bump timeout for client")

178

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

179

help="Start checker for client")

180

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

181

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

182

159

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

183

160

help="Check if client is enabled")

184

161

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

185

162

help="Remove client")

186

163

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

187

164

help="Set checker command for client")

188

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

165

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

189

166

help="Set timeout for client")

190

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

167

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

191

168

help="Set extended timeout for client")

192

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

169

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

193

170

help="Set checker interval for client")

194

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

195

default=None, dest="approved_by_default",

196

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

197

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

198

dest="approved_by_default",

199

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

200

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,

201

181

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

202

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

182

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

203

183

help="Set duration of one client approval")

204

184

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

205

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

185

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

186

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

206

187

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

207

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

208

help="Approve any current client request")

209

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

210

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

211

198

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

212

options = parser.parse_args()

213

214

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

236

>>> # Must start with "P":

237

>>> rfc3339_duration_to_delta("1D")

238

Traceback (most recent call last):

239

...

240

ValueError: Invalid RFC 3339 duration: "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: "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: "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: "P1D2W"

256

>>> rfc3339_duration_to_delta("P2W2H")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: "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: \"{}\""

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

215

410

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

216

411

if options.verbose and has_actions(options):

217

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

218

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

219

416

if options.all and not has_actions(options):

220

417

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

221

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

222

439

try:

223

bus = dbus.SystemBus()

224

mandos_dbus_objc = bus.get_object(busname, server_path)

225

except dbus.exceptions.DBusException:

226

print("Could not connect to Mandos server",

227

file=sys.stderr)

440

yield

441

except dbus.exceptions.DBusException as e:

442

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

228

443

sys.exit(1)

229

230

mandos_serv = dbus.Interface(mandos_dbus_objc,

231

dbus_interface = server_interface)

232

233

#block stderr since dbus library prints to stderr

234

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

235

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

236

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

237

os.close(null)

238

try:

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(command.IsEnabled())

481

482

if options.approve:

483

commands.append(command.Approve())

484

485

if options.deny:

486

commands.append(command.Deny())

487

488

if options.remove:

489

commands.append(command.Remove())

490

491

if options.dump_json:

492

commands.append(command.DumpJSON())

493

494

if options.enable:

495

commands.append(command.Enable())

496

497

if options.disable:

498

commands.append(command.Disable())

499

500

if options.bump_timeout:

501

commands.append(command.BumpTimeout())

502

503

if options.start_checker:

504

commands.append(command.StartChecker())

505

506

if options.stop_checker:

507

commands.append(command.StopChecker())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(command.ApproveByDefault())

512

else:

513

commands.append(command.DenyByDefault())

514

515

if options.checker is not None:

516

commands.append(command.SetChecker(options.checker))

517

518

if options.host is not None:

519

commands.append(command.SetHost(options.host))

520

521

if options.secret is not None:

522

commands.append(command.SetSecret(options.secret))

523

524

if options.timeout is not None:

525

commands.append(command.SetTimeout(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

command.SetExtendedTimeout(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(command.SetInterval(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(

536

command.SetApprovalDelay(options.approval_delay))

537

538

if options.approval_duration is not None:

539

commands.append(

540

command.SetApprovalDuration(options.approval_duration))

541

542

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

543

# optionally verbosely

544

if not commands:

545

commands.append(command.PrintTable(verbose=options.verbose))

546

547

return commands

548

549

550

class command(object):

551

"""A namespace for command classes"""

552

553

class Base(object):

554

"""Abstract base class for commands"""

555

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

556

"""Normal commands should implement run_on_one_client(),

557

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

558

override this run() method instead.

559

"""

560

self.mandos = mandos

561

for clientpath, properties in clients.items():

562

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

563

dbus_busname, str(clientpath))

564

client = bus.get_object(dbus_busname, clientpath)

565

self.run_on_one_client(client, properties)

566

567

568

class IsEnabled(Base):

569

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

570

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

571

if self.is_enabled(client, properties):

572

sys.exit(0)

573

sys.exit(1)

574

def is_enabled(self, client, properties):

575

return properties["Enabled"]

576

577

578

class Approve(Base):

579

def run_on_one_client(self, client, properties):

580

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

581

client.__dbus_object_path__,

582

client_dbus_interface)

583

client.Approve(dbus.Boolean(True),

584

dbus_interface=client_dbus_interface)

585

586

587

class Deny(Base):

588

def run_on_one_client(self, client, properties):

589

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

590

client.__dbus_object_path__,

591

client_dbus_interface)

592

client.Approve(dbus.Boolean(False),

593

dbus_interface=client_dbus_interface)

594

595

596

class Remove(Base):

597

def run_on_one_client(self, client, properties):

598

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

599

dbus_busname, server_dbus_path,

600

server_dbus_interface,

601

str(client.__dbus_object_path__))

602

self.mandos.RemoveClient(client.__dbus_object_path__)

603

604

605

class Output(Base):

606

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

607

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

608

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

609

"Fingerprint", "CheckerRunning",

610

"LastEnabled", "ApprovalPending",

611

"ApprovedByDefault", "LastApprovalRequest",

612

"ApprovalDelay", "ApprovalDuration",

613

"Checker", "ExtendedTimeout", "Expires",

614

"LastCheckerStatus")

615

616

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

617

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

618

619

def output(self, clients):

620

raise NotImplementedError()

621

622

623

class DumpJSON(Output):

624

def output(self, clients):

625

data = {client["Name"]:

626

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

627

for key in self.all_keywords}

628

for client in clients}

629

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

630

631

@staticmethod

632

def dbus_boolean_to_bool(value):

633

if isinstance(value, dbus.Boolean):

634

value = bool(value)

635

return value

636

637

638

class PrintTable(Output):

639

def __init__(self, verbose=False):

640

self.verbose = verbose

641

642

def output(self, clients):

643

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

644

"LastCheckedOK")

645

keywords = default_keywords

646

if self.verbose:

647

keywords = self.all_keywords

648

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

649

650

class TableOfClients(object):

651

tableheaders = {

652

"Name": "Name",

653

"Enabled": "Enabled",

654

"Timeout": "Timeout",

655

"LastCheckedOK": "Last Successful Check",

656

"LastApprovalRequest": "Last Approval Request",

657

"Created": "Created",

658

"Interval": "Interval",

659

"Host": "Host",

660

"Fingerprint": "Fingerprint",

661

"KeyID": "Key ID",

662

"CheckerRunning": "Check Is Running",

663

"LastEnabled": "Last Enabled",

664

"ApprovalPending": "Approval Is Pending",

665

"ApprovedByDefault": "Approved By Default",

666

"ApprovalDelay": "Approval Delay",

667

"ApprovalDuration": "Approval Duration",

668

"Checker": "Checker",

669

"ExtendedTimeout": "Extended Timeout",

670

"Expires": "Expires",

671

"LastCheckerStatus": "Last Checker Status",

672

}

673

674

def __init__(self, clients, keywords):

675

self.clients = clients

676

self.keywords = keywords

677

678

def __str__(self):

679

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

680

681

if sys.version_info.major == 2:

682

__unicode__ = __str__

683

def __str__(self):

684

return str(self).encode(

685

locale.getpreferredencoding())

686

687

def rows(self):

688

format_string = self.row_formatting_string()

689

rows = [self.header_line(format_string)]

690

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

691

for client in self.clients)

692

return rows

693

694

def row_formatting_string(self):

695

"Format string used to format table rows"

696

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

697

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

698

*(len(self.string_from_client(client,

699

key))

700

for client in self.clients)),

701

key=key)

702

for key in self.keywords)

703

704

def string_from_client(self, client, key):

705

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

706

707

@classmethod

708

def valuetostring(cls, value, keyword):

709

if isinstance(value, dbus.Boolean):

710

return "Yes" if value else "No"

711

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

712

"ApprovalDuration", "ExtendedTimeout"):

713

return cls.milliseconds_to_string(value)

714

return str(value)

715

716

def header_line(self, format_string):

717

return format_string.format(**self.tableheaders)

718

719

def client_line(self, client, format_string):

720

return format_string.format(

721

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

722

for key in self.keywords})

723

724

@staticmethod

725

def milliseconds_to_string(ms):

726

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

727

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

728

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

729

if td.days else "",

730

hours=td.seconds // 3600,

731

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

732

seconds=td.seconds % 60))

733

734

735

class Property(Base):

736

"Abstract class for Actions for setting one client property"

737

738

def run_on_one_client(self, client, properties):

739

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

740

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

741

client.__dbus_object_path__,

742

dbus.PROPERTIES_IFACE, client_dbus_interface,

743

self.propname, self.value_to_set

744

if not isinstance(self.value_to_set,

745

dbus.Boolean)

746

else bool(self.value_to_set))

747

client.Set(client_dbus_interface, self.propname,

748

self.value_to_set,

749

dbus_interface=dbus.PROPERTIES_IFACE)

750

751

@property

752

def propname(self):

753

raise NotImplementedError()

754

755

756

class Enable(Property):

757

propname = "Enabled"

758

value_to_set = dbus.Boolean(True)

759

760

761

class Disable(Property):

762

propname = "Enabled"

763

value_to_set = dbus.Boolean(False)

764

765

766

class BumpTimeout(Property):

767

propname = "LastCheckedOK"

768

value_to_set = ""

769

770

771

class StartChecker(Property):

772

propname = "CheckerRunning"

773

value_to_set = dbus.Boolean(True)

774

775

776

class StopChecker(Property):

777

propname = "CheckerRunning"

778

value_to_set = dbus.Boolean(False)

779

780

781

class ApproveByDefault(Property):

782

propname = "ApprovedByDefault"

783

value_to_set = dbus.Boolean(True)

784

785

786

class DenyByDefault(Property):

787

propname = "ApprovedByDefault"

788

value_to_set = dbus.Boolean(False)

789

790

791

class PropertyValue(Property):

792

"Abstract class for Property recieving a value as argument"

793

def __init__(self, value):

794

self.value_to_set = value

795

796

797

class SetChecker(PropertyValue):

798

propname = "Checker"

799

800

801

class SetHost(PropertyValue):

802

propname = "Host"

803

804

805

class SetSecret(PropertyValue):

806

propname = "Secret"

807

808

@property

809

def value_to_set(self):

810

return self._vts

811

812

@value_to_set.setter

813

def value_to_set(self, value):

814

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

815

self._vts = value.read()

816

value.close()

817

818

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

822

823

@property

824

def value_to_set(self):

825

return self._vts

826

827

@value_to_set.setter

828

def value_to_set(self, value):

829

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

830

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

831

832

833

class SetTimeout(MillisecondsPropertyValueArgument):

834

propname = "Timeout"

835

836

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

838

propname = "ExtendedTimeout"

839

840

841

class SetInterval(MillisecondsPropertyValueArgument):

842

propname = "Interval"

843

844

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

846

propname = "ApprovalDelay"

847

848

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

850

propname = "ApprovalDuration"

851

852

853

854

class TestCaseWithAssertLogs(unittest.TestCase):

855

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

856

857

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

858

@contextlib.contextmanager

859

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

860

capturing_handler = self.CapturingLevelHandler(level)

861

old_level = logger.level

862

old_propagate = logger.propagate

863

logger.addHandler(capturing_handler)

864

logger.setLevel(level)

865

logger.propagate = False

866

try:

867

yield capturing_handler.watcher

868

finally:

869

logger.propagate = old_propagate

870

logger.removeHandler(capturing_handler)

871

logger.setLevel(old_level)

872

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

873

0)

874

875

class CapturingLevelHandler(logging.Handler):

876

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

877

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

878

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

879

def emit(self, record):

880

self.watcher.records.append(record)

881

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

882

883

LoggingWatcher = collections.namedtuple("LoggingWatcher",

884

("records",

885

"output"))

886

887

888

class Test_string_to_delta(TestCaseWithAssertLogs):

889

# Just test basic RFC 3339 functionality here, the doc string for

890

# rfc3339_duration_to_delta() already has more comprehensive

891

# tests, which is run by doctest.

892

893

def test_rfc3339_zero_seconds(self):

894

self.assertEqual(datetime.timedelta(),

895

string_to_delta("PT0S"))

896

897

def test_rfc3339_zero_days(self):

898

self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))

899

900

def test_rfc3339_one_second(self):

901

self.assertEqual(datetime.timedelta(0, 1),

902

string_to_delta("PT1S"))

903

904

def test_rfc3339_two_hours(self):

905

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

906

string_to_delta("PT2H"))

907

908

def test_falls_back_to_pre_1_6_1_with_warning(self):

909

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

910

value = string_to_delta("2h")

911

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

912

913

914

class Test_check_option_syntax(unittest.TestCase):

915

def setUp(self):

916

self.parser = argparse.ArgumentParser()

917

add_command_line_options(self.parser)

918

919

def test_actions_requires_client_or_all(self):

920

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

921

options = self.parser.parse_args()

922

setattr(options, action, value)

923

with self.assertParseError():

924

self.check_option_syntax(options)

925

926

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

927

# check_option_syntax()

928

actions = {

929

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

930

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

931

# never used

932

"enable": True,

933

"disable": True,

934

"bump_timeout": True,

935

"start_checker": True,

936

"stop_checker": True,

937

"is_enabled": True,

938

"remove": True,

939

"checker": "x",

940

"timeout": datetime.timedelta(),

941

"extended_timeout": datetime.timedelta(),

942

"interval": datetime.timedelta(),

943

"approved_by_default": True,

944

"approval_delay": datetime.timedelta(),

945

"approval_duration": datetime.timedelta(),

946

"host": "x",

947

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

948

"approve": True,

949

"deny": True,

950

}

951

952

@contextlib.contextmanager

953

def assertParseError(self):

954

with self.assertRaises(SystemExit) as e:

955

with self.redirect_stderr_to_devnull():

956

yield

957

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

958

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

959

# /argparse.html#exiting-methods

960

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

961

962

@staticmethod

963

@contextlib.contextmanager

964

def redirect_stderr_to_devnull():

965

old_stderr = sys.stderr

966

with contextlib.closing(open(os.devnull, "w")) as null:

967

sys.stderr = null

968

try:

969

yield

970

finally:

971

sys.stderr = old_stderr

972

973

def check_option_syntax(self, options):

974

check_option_syntax(self.parser, options)

975

976

def test_actions_all_conflicts_with_verbose(self):

977

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

978

options = self.parser.parse_args()

979

setattr(options, action, value)

980

options.all = True

981

options.verbose = True

982

with self.assertParseError():

983

self.check_option_syntax(options)

984

985

def test_actions_with_client_conflicts_with_verbose(self):

986

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

987

options = self.parser.parse_args()

988

setattr(options, action, value)

989

options.verbose = True

990

options.client = ["foo"]

991

with self.assertParseError():

992

self.check_option_syntax(options)

993

994

def test_dump_json_conflicts_with_verbose(self):

995

options = self.parser.parse_args()

996

options.dump_json = True

997

options.verbose = True

998

with self.assertParseError():

999

self.check_option_syntax(options)

1000

1001

def test_dump_json_conflicts_with_action(self):

1002

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

1003

options = self.parser.parse_args()

1004

setattr(options, action, value)

1005

options.dump_json = True

1006

with self.assertParseError():

1007

self.check_option_syntax(options)

1008

1009

def test_all_can_not_be_alone(self):

1010

options = self.parser.parse_args()

1011

options.all = True

1012

with self.assertParseError():

1013

self.check_option_syntax(options)

1014

1015

def test_all_is_ok_with_any_action(self):

1016

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

1017

options = self.parser.parse_args()

1018

setattr(options, action, value)

1019

options.all = True

1020

self.check_option_syntax(options)

1021

1022

def test_any_action_is_ok_with_one_client(self):

1023

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

1024

options = self.parser.parse_args()

1025

setattr(options, action, value)

1026

options.client = ["foo"]

1027

self.check_option_syntax(options)

1028

1029

def test_one_client_with_all_actions_except_is_enabled(self):

1030

options = self.parser.parse_args()

1031

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

1032

if action == "is_enabled":

1033

continue

1034

setattr(options, action, value)

1035

options.client = ["foo"]

1036

self.check_option_syntax(options)

1037

1038

def test_two_clients_with_all_actions_except_is_enabled(self):

1039

options = self.parser.parse_args()

1040

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

1041

if action == "is_enabled":

1042

continue

1043

setattr(options, action, value)

1044

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

1045

self.check_option_syntax(options)

1046

1047

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1048

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

1049

if action == "is_enabled":

1050

continue

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

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

1054

self.check_option_syntax(options)

1055

1056

def test_is_enabled_fails_without_client(self):

1057

options = self.parser.parse_args()

1058

options.is_enabled = True

1059

with self.assertParseError():

1060

self.check_option_syntax(options)

1061

1062

def test_is_enabled_fails_with_two_clients(self):

1063

options = self.parser.parse_args()

1064

options.is_enabled = True

1065

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

1066

with self.assertParseError():

1067

self.check_option_syntax(options)

1068

1069

def test_remove_can_only_be_combined_with_action_deny(self):

1070

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

1071

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

1072

continue

1073

options = self.parser.parse_args()

1074

setattr(options, action, value)

1075

options.all = True

1076

options.remove = True

1077

with self.assertParseError():

1078

self.check_option_syntax(options)

1079

1080

1081

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1082

def test_calls_and_returns_get_object_on_bus(self):

1083

class MockBus(object):

1084

called = False

1085

def get_object(mockbus_self, busname, dbus_path):

1086

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

1087

# this MockBus instance is in "mockbus_self".

1088

self.assertEqual(dbus_busname, busname)

1089

self.assertEqual(server_dbus_path, dbus_path)

1090

mockbus_self.called = True

1091

return mockbus_self

1092

1093

mockbus = get_mandos_dbus_object(bus=MockBus())

1094

self.assertIsInstance(mockbus, MockBus)

1095

self.assertTrue(mockbus.called)

1096

1097

def test_logs_and_exits_on_dbus_error(self):

1098

class MockBusFailing(object):

1099

def get_object(self, busname, dbus_path):

1100

raise dbus.exceptions.DBusException("Test")

1101

1102

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

1103

with self.assertRaises(SystemExit) as e:

1104

bus = get_mandos_dbus_object(bus=MockBusFailing())

1105

1106

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

1107

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

1108

else:

1109

self.assertIsNotNone(e.exception.code)

1110

1111

1112

class Test_get_managed_objects(TestCaseWithAssertLogs):

1113

def test_calls_and_returns_GetManagedObjects(self):

1114

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

1115

class MockObjectManager(object):

1116

def GetManagedObjects(self):

1117

return managed_objects

1118

retval = get_managed_objects(MockObjectManager())

1119

self.assertDictEqual(managed_objects, retval)

1120

1121

def test_logs_and_exits_on_dbus_error(self):

1122

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

1123

1124

class MockObjectManagerFailing(object):

1125

def GetManagedObjects(self):

1126

dbus_logger.error("Test")

1127

raise dbus.exceptions.DBusException("Test")

1128

1129

class CountingHandler(logging.Handler):

1130

count = 0

1131

def emit(self, record):

1132

self.count += 1

1133

1134

counting_handler = CountingHandler()

1135

1136

dbus_logger.addHandler(counting_handler)

1137

239

1138

try:

240

mandos_clients = mandos_serv.GetAllClientsWithProperties()

1139

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

1140

with self.assertRaises(SystemExit) as e:

1141

get_managed_objects(MockObjectManagerFailing())

241

1142

finally:

242

#restore stderr

243

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

244

os.close(stderrcopy)

245

except dbus.exceptions.DBusException, e:

246

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

247

file=sys.stderr)

248

sys.exit(1)

249

250

# Compile dict of (clients: properties) to process

251

clients={}

252

253

if options.all or not options.client:

254

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

255

for path, properties in

256

mandos_clients.iteritems())

257

else:

258

for name in options.client:

259

for path, client in mandos_clients.iteritems():

260

if client["Name"] == name:

261

client_objc = bus.get_object(busname, path)

262

clients[client_objc] = client

263

break

264

else:

265

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

266

file=sys.stderr)

267

sys.exit(1)

268

269

if not has_actions(options) and clients:

270

if options.verbose:

271

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

272

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

273

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

274

"LastEnabled", "ApprovalPending",

275

"ApprovedByDefault",

276

"LastApprovalRequest", "ApprovalDelay",

277

"ApprovalDuration", "Checker",

278

"ExtendedTimeout")

279

else:

280

keywords = defaultkeywords

281

282

print_clients(clients.values(), keywords)

283

else:

284

# Process each client in the list by all selected options

285

for client in clients:

286

if options.remove:

287

mandos_serv.RemoveClient(client.__dbus_object_path__)

288

if options.enable:

289

client.Enable(dbus_interface=client_interface)

290

if options.disable:

291

client.Disable(dbus_interface=client_interface)

292

if options.bump_timeout:

293

client.CheckedOK(dbus_interface=client_interface)

294

if options.start_checker:

295

client.StartChecker(dbus_interface=client_interface)

296

if options.stop_checker:

297

client.StopChecker(dbus_interface=client_interface)

298

if options.is_enabled:

299

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

300

"Enabled",

301

dbus_interface=

302

dbus.PROPERTIES_IFACE)

303

else 1)

304

if options.checker:

305

client.Set(client_interface, "Checker",

306

options.checker,

307

dbus_interface=dbus.PROPERTIES_IFACE)

308

if options.host:

309

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

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

if options.interval:

312

client.Set(client_interface, "Interval",

313

timedelta_to_milliseconds

314

(string_to_delta(options.interval)),

315

dbus_interface=dbus.PROPERTIES_IFACE)

316

if options.approval_delay:

317

client.Set(client_interface, "ApprovalDelay",

318

timedelta_to_milliseconds

319

(string_to_delta(options.

320

approval_delay)),

321

dbus_interface=dbus.PROPERTIES_IFACE)

322

if options.approval_duration:

323

client.Set(client_interface, "ApprovalDuration",

324

timedelta_to_milliseconds

325

(string_to_delta(options.

326

approval_duration)),

327

dbus_interface=dbus.PROPERTIES_IFACE)

328

if options.timeout:

329

client.Set(client_interface, "Timeout",

330

timedelta_to_milliseconds

331

(string_to_delta(options.timeout)),

332

dbus_interface=dbus.PROPERTIES_IFACE)

333

if options.extended_timeout:

334

client.Set(client_interface, "ExtendedTimeout",

335

timedelta_to_milliseconds

336

(string_to_delta(options.extended_timeout)),

337

dbus_interface=dbus.PROPERTIES_IFACE)

338

if options.secret:

339

client.Set(client_interface, "Secret",

340

dbus.ByteArray(open(options.secret,

341

"rb").read()),

342

dbus_interface=dbus.PROPERTIES_IFACE)

343

if options.approved_by_default is not None:

344

client.Set(client_interface, "ApprovedByDefault",

345

dbus.Boolean(options

346

.approved_by_default),

347

dbus_interface=dbus.PROPERTIES_IFACE)

348

if options.approve:

349

client.Approve(dbus.Boolean(True),

350

dbus_interface=client_interface)

351

elif options.deny:

352

client.Approve(dbus.Boolean(False),

353

dbus_interface=client_interface)

1143

dbus_logger.removeFilter(counting_handler)

1144

1145

# Make sure the dbus logger was suppressed

1146

self.assertEqual(0, counting_handler.count)

1147

1148

# Test that the dbus_logger still works

1149

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

1150

dbus_logger.error("Test")

1151

1152

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

1153

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

1154

else:

1155

self.assertIsNotNone(e.exception.code)

1156

1157

1158

class Test_commands_from_options(unittest.TestCase):

1159

def setUp(self):

1160

self.parser = argparse.ArgumentParser()

1161

add_command_line_options(self.parser)

1162

1163

def test_is_enabled(self):

1164

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

1165

command.IsEnabled)

1166

1167

def assert_command_from_args(self, args, command_cls,

1168

**cmd_attrs):

1169

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

1170

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

1171

options = self.parser.parse_args(args)

1172

check_option_syntax(self.parser, options)

1173

commands = commands_from_options(options)

1174

self.assertEqual(1, len(commands))

1175

command = commands[0]

1176

self.assertIsInstance(command, command_cls)

1177

for key, value in cmd_attrs.items():

1178

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

1179

1180

def test_is_enabled_short(self):

1181

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

1182

command.IsEnabled)

1183

1184

def test_approve(self):

1185

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

1186

command.Approve)

1187

1188

def test_approve_short(self):

1189

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

1190

1191

def test_deny(self):

1192

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

1193

1194

def test_deny_short(self):

1195

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

1196

1197

def test_remove(self):

1198

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

1199

command.Remove)

1200

1201

def test_deny_before_remove(self):

1202

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

1203

"foo"])

1204

check_option_syntax(self.parser, options)

1205

commands = commands_from_options(options)

1206

self.assertEqual(2, len(commands))

1207

self.assertIsInstance(commands[0], command.Deny)

1208

self.assertIsInstance(commands[1], command.Remove)

1209

1210

def test_deny_before_remove_reversed(self):

1211

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

1212

"--all"])

1213

check_option_syntax(self.parser, options)

1214

commands = commands_from_options(options)

1215

self.assertEqual(2, len(commands))

1216

self.assertIsInstance(commands[0], command.Deny)

1217

self.assertIsInstance(commands[1], command.Remove)

1218

1219

def test_remove_short(self):

1220

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

1221

1222

def test_dump_json(self):

1223

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

1224

command.DumpJSON)

1225

1226

def test_enable(self):

1227

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

1228

command.Enable)

1229

1230

def test_enable_short(self):

1231

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

1232

1233

def test_disable(self):

1234

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

1235

command.Disable)

1236

1237

def test_disable_short(self):

1238

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

1239

1240

def test_bump_timeout(self):

1241

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

1242

command.BumpTimeout)

1243

1244

def test_bump_timeout_short(self):

1245

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

1246

command.BumpTimeout)

1247

1248

def test_start_checker(self):

1249

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

1250

command.StartChecker)

1251

1252

def test_stop_checker(self):

1253

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

1254

command.StopChecker)

1255

1256

def test_approve_by_default(self):

1257

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

1258

command.ApproveByDefault)

1259

1260

def test_deny_by_default(self):

1261

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

1262

command.DenyByDefault)

1263

1264

def test_checker(self):

1265

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

1266

command.SetChecker,

1267

value_to_set=":")

1268

1269

def test_checker_empty(self):

1270

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

1271

command.SetChecker,

1272

value_to_set="")

1273

1274

def test_checker_short(self):

1275

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

1276

command.SetChecker,

1277

value_to_set=":")

1278

1279

def test_host(self):

1280

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

1281

"foo"], command.SetHost,

1282

value_to_set="foo.example.org")

1283

1284

def test_host_short(self):

1285

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

1286

"foo"], command.SetHost,

1287

value_to_set="foo.example.org")

1288

1289

def test_secret_devnull(self):

1290

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

1291

"foo"], command.SetSecret,

1292

value_to_set=b"")

1293

1294

def test_secret_tempfile(self):

1295

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

1296

value = b"secret\0xyzzy\nbar"

1297

f.write(value)

1298

f.seek(0)

1299

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

1300

"foo"], command.SetSecret,

1301

value_to_set=value)

1302

1303

def test_secret_devnull_short(self):

1304

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

1305

command.SetSecret,

1306

value_to_set=b"")

1307

1308

def test_secret_tempfile_short(self):

1309

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

1310

value = b"secret\0xyzzy\nbar"

1311

f.write(value)

1312

f.seek(0)

1313

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

1314

command.SetSecret,

1315

value_to_set=value)

1316

1317

def test_timeout(self):

1318

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

1319

command.SetTimeout,

1320

value_to_set=300000)

1321

1322

def test_timeout_short(self):

1323

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

1324

command.SetTimeout,

1325

value_to_set=300000)

1326

1327

def test_extended_timeout(self):

1328

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

1329

"foo"],

1330

command.SetExtendedTimeout,

1331

value_to_set=900000)

1332

1333

def test_interval(self):

1334

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

1335

command.SetInterval,

1336

value_to_set=120000)

1337

1338

def test_interval_short(self):

1339

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

1340

command.SetInterval,

1341

value_to_set=120000)

1342

1343

def test_approval_delay(self):

1344

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

1345

"foo"],

1346

command.SetApprovalDelay,

1347

value_to_set=30000)

1348

1349

def test_approval_duration(self):

1350

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

1351

"foo"],

1352

command.SetApprovalDuration,

1353

value_to_set=1000)

1354

1355

def test_print_table(self):

1356

self.assert_command_from_args([], command.PrintTable,

1357

verbose=False)

1358

1359

def test_print_table_verbose(self):

1360

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

1361

command.PrintTable,

1362

verbose=True)

1363

1364

def test_print_table_verbose_short(self):

1365

self.assert_command_from_args(["-v"], command.PrintTable,

1366

verbose=True)

1367

1368

1369

class TestCommand(unittest.TestCase):

1370

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

1371

1372

def setUp(self):

1373

testcase = self

1374

class MockClient(object):

1375

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

1376

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

1377

self.attributes = attributes

1378

self.attributes["Name"] = name

1379

self.calls = []

1380

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

1381

testcase.assertEqual(client_dbus_interface, interface)

1382

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1383

dbus_interface)

1384

self.attributes[propname] = value

1385

def Approve(self, approve, dbus_interface):

1386

testcase.assertEqual(client_dbus_interface,

1387

dbus_interface)

1388

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

1389

dbus_interface)))

1390

self.client = MockClient(

1391

"foo",

1392

KeyID=("92ed150794387c03ce684574b1139a65"

1393

"94a34f895daaaf09fd8ea90a27cddb12"),

1394

Secret=b"secret",

1395

Host="foo.example.org",

1396

Enabled=dbus.Boolean(True),

1397

Timeout=300000,

1398

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

1399

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

1400

Interval=120000,

1401

Fingerprint=("778827225BA7DE539C5A"

1402

"7CFA59CFF7CDBD9A5920"),

1403

CheckerRunning=dbus.Boolean(False),

1404

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

1405

ApprovalPending=dbus.Boolean(False),

1406

ApprovedByDefault=dbus.Boolean(True),

1407

LastApprovalRequest="",

1408

ApprovalDelay=0,

1409

ApprovalDuration=1000,

1410

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

1411

ExtendedTimeout=900000,

1412

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

1413

LastCheckerStatus=0)

1414

self.other_client = MockClient(

1415

"barbar",

1416

KeyID=("0558568eedd67d622f5c83b35a115f79"

1417

"6ab612cff5ad227247e46c2b020f441c"),

1418

Secret=b"secretbar",

1419

Host="192.0.2.3",

1420

Enabled=dbus.Boolean(True),

1421

Timeout=300000,

1422

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

1423

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

1424

Interval=120000,

1425

Fingerprint=("3E393AEAEFB84C7E89E2"

1426

"F547B3A107558FCA3A27"),

1427

CheckerRunning=dbus.Boolean(True),

1428

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

1429

ApprovalPending=dbus.Boolean(False),

1430

ApprovedByDefault=dbus.Boolean(False),

1431

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

1432

ApprovalDelay=30000,

1433

ApprovalDuration=93785000,

1434

Checker=":",

1435

ExtendedTimeout=900000,

1436

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

1437

LastCheckerStatus=-2)

1438

self.clients = collections.OrderedDict(

1439

[

1440

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

1441

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

1442

])

1443

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

1444

1445

@property

1446

def bus(self):

1447

class Bus(object):

1448

@staticmethod

1449

def get_object(client_bus_name, path):

1450

self.assertEqual(dbus_busname, client_bus_name)

1451

return {

1452

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

1453

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

1454

"/clients/foo": self.client,

1455

"/clients/barbar": self.other_client,

1456

}[path]

1457

return Bus()

1458

1459

1460

class TestBaseCommands(TestCommand):

1461

1462

def test_IsEnabled_exits_successfully(self):

1463

with self.assertRaises(SystemExit) as e:

1464

command.IsEnabled().run(self.one_client)

1465

if e.exception.code is not None:

1466

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

1467

else:

1468

self.assertIsNone(e.exception.code)

1469

1470

def test_IsEnabled_exits_with_failure(self):

1471

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

1472

with self.assertRaises(SystemExit) as e:

1473

command.IsEnabled().run(self.one_client)

1474

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

1475

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

1476

else:

1477

self.assertIsNotNone(e.exception.code)

1478

1479

def test_Approve(self):

1480

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

1481

for clientpath in self.clients:

1482

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

1483

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

1484

client.calls)

1485

1486

def test_Deny(self):

1487

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

1488

for clientpath in self.clients:

1489

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

1490

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

1491

client.calls)

1492

1493

def test_Remove(self):

1494

class MockMandos(object):

1495

def __init__(self):

1496

self.calls = []

1497

def RemoveClient(self, dbus_path):

1498

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

1499

mandos = MockMandos()

1500

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

1501

for clientpath in self.clients:

1502

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

1503

mandos.calls)

1504

1505

expected_json = {

1506

"foo": {

1507

"Name": "foo",

1508

"KeyID": ("92ed150794387c03ce684574b1139a65"

1509

"94a34f895daaaf09fd8ea90a27cddb12"),

1510

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

1511

"Enabled": True,

1512

"Timeout": 300000,

1513

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

1514

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

1515

"Interval": 120000,

1516

"Fingerprint": ("778827225BA7DE539C5A"

1517

"7CFA59CFF7CDBD9A5920"),

1518

"CheckerRunning": False,

1519

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

1520

"ApprovalPending": False,

1521

"ApprovedByDefault": True,

1522

"LastApprovalRequest": "",

1523

"ApprovalDelay": 0,

1524

"ApprovalDuration": 1000,

1525

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

1526

"ExtendedTimeout": 900000,

1527

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

1528

"LastCheckerStatus": 0,

1529

},

1530

"barbar": {

1531

"Name": "barbar",

1532

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1533

"6ab612cff5ad227247e46c2b020f441c"),

1534

"Host": "192.0.2.3",

1535

"Enabled": True,

1536

"Timeout": 300000,

1537

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

1538

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

1539

"Interval": 120000,

1540

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1541

"F547B3A107558FCA3A27"),

1542

"CheckerRunning": True,

1543

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

1544

"ApprovalPending": False,

1545

"ApprovedByDefault": False,

1546

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

1547

"ApprovalDelay": 30000,

1548

"ApprovalDuration": 93785000,

1549

"Checker": ":",

1550

"ExtendedTimeout": 900000,

1551

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

1552

"LastCheckerStatus": -2,

1553

},

1554

}

1555

1556

def test_DumpJSON_normal(self):

1557

output = command.DumpJSON().output(self.clients.values())

1558

json_data = json.loads(output)

1559

self.assertDictEqual(self.expected_json, json_data)

1560

1561

def test_DumpJSON_one_client(self):

1562

output = command.DumpJSON().output(self.one_client.values())

1563

json_data = json.loads(output)

1564

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

1565

self.assertDictEqual(expected_json, json_data)

1566

1567

def test_PrintTable_normal(self):

1568

output = command.PrintTable().output(self.clients.values())

1569

expected_output = "\n".join((

1570

"Name Enabled Timeout Last Successful Check",

1571

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

1572

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

1573

))

1574

self.assertEqual(expected_output, output)

1575

1576

def test_PrintTable_verbose(self):

1577

output = command.PrintTable(verbose=True).output(

1578

self.clients.values())

1579

columns = (

1580

(

1581

"Name ",

1582

"foo ",

1583

"barbar ",

1584

),(

1585

"Enabled ",

1586

"Yes ",

1587

"Yes ",

1588

),(

1589

"Timeout ",

1590

"00:05:00 ",

1591

"00:05:00 ",

1592

),(

1593

"Last Successful Check ",

1594

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

1595

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

1596

),(

1597

"Created ",

1598

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

1599

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

1600

),(

1601

"Interval ",

1602

"00:02:00 ",

1603

"00:02:00 ",

1604

),(

1605

"Host ",

1606

"foo.example.org ",

1607

"192.0.2.3 ",

1608

),(

1609

("Key ID "

1610

" "),

1611

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1612

"ea90a27cddb12 "),

1613

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1614

"46c2b020f441c "),

1615

),(

1616

"Fingerprint ",

1617

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1618

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1619

),(

1620

"Check Is Running ",

1621

"No ",

1622

"Yes ",

1623

),(

1624

"Last Enabled ",

1625

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

1626

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

1627

),(

1628

"Approval Is Pending ",

1629

"No ",

1630

"No ",

1631

),(

1632

"Approved By Default ",

1633

"Yes ",

1634

"No ",

1635

),(

1636

"Last Approval Request ",

1637

" ",

1638

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

1639

),(

1640

"Approval Delay ",

1641

"00:00:00 ",

1642

"00:00:30 ",

1643

),(

1644

"Approval Duration ",

1645

"00:00:01 ",

1646

"1T02:03:05 ",

1647

),(

1648

"Checker ",

1649

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

1650

": ",

1651

),(

1652

"Extended Timeout ",

1653

"00:15:00 ",

1654

"00:15:00 ",

1655

),(

1656

"Expires ",

1657

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

1658

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

1659

),(

1660

"Last Checker Status",

1661

"0 ",

1662

"-2 ",

1663

)

1664

)

1665

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

1666

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

1667

for rows in columns)

1668

for line in range(num_lines))

1669

self.assertEqual(expected_output, output)

1670

1671

def test_PrintTable_one_client(self):

1672

output = command.PrintTable().output(self.one_client.values())

1673

expected_output = "\n".join((

1674

"Name Enabled Timeout Last Successful Check",

1675

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

1676

))

1677

self.assertEqual(expected_output, output)

1678

1679

1680

class TestPropertyCmd(TestCommand):

1681

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

1682

def runTest(self):

1683

if not hasattr(self, "command"):

1684

return

1685

values_to_get = getattr(self, "values_to_get",

1686

self.values_to_set)

1687

for value_to_set, value_to_get in zip(self.values_to_set,

1688

values_to_get):

1689

for clientpath in self.clients:

1690

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

1691

old_value = client.attributes[self.propname]

1692

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

1693

self.run_command(value_to_set, self.clients)

1694

for clientpath in self.clients:

1695

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

1696

value = client.attributes[self.propname]

1697

self.assertNotIsInstance(value, self.Unique)

1698

self.assertEqual(value_to_get, value)

1699

1700

class Unique(object):

1701

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

1702

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

1703

1704

def run_command(self, value, clients):

1705

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

1706

1707

1708

class TestEnableCmd(TestPropertyCmd):

1709

command = command.Enable

1710

propname = "Enabled"

1711

values_to_set = [dbus.Boolean(True)]

1712

1713

1714

class TestDisableCmd(TestPropertyCmd):

1715

command = command.Disable

1716

propname = "Enabled"

1717

values_to_set = [dbus.Boolean(False)]

1718

1719

1720

class TestBumpTimeoutCmd(TestPropertyCmd):

1721

command = command.BumpTimeout

1722

propname = "LastCheckedOK"

1723

values_to_set = [""]

1724

1725

1726

class TestStartCheckerCmd(TestPropertyCmd):

1727

command = command.StartChecker

1728

propname = "CheckerRunning"

1729

values_to_set = [dbus.Boolean(True)]

1730

1731

1732

class TestStopCheckerCmd(TestPropertyCmd):

1733

command = command.StopChecker

1734

propname = "CheckerRunning"

1735

values_to_set = [dbus.Boolean(False)]

1736

1737

1738

class TestApproveByDefaultCmd(TestPropertyCmd):

1739

command = command.ApproveByDefault

1740

propname = "ApprovedByDefault"

1741

values_to_set = [dbus.Boolean(True)]

1742

1743

1744

class TestDenyByDefaultCmd(TestPropertyCmd):

1745

command = command.DenyByDefault

1746

propname = "ApprovedByDefault"

1747

values_to_set = [dbus.Boolean(False)]

1748

1749

1750

class TestPropertyValueCmd(TestPropertyCmd):

1751

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

1752

1753

def runTest(self):

1754

if type(self) is TestPropertyValueCmd:

1755

return

1756

return super(TestPropertyValueCmd, self).runTest()

1757

1758

def run_command(self, value, clients):

1759

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

1760

1761

1762

class TestSetCheckerCmd(TestPropertyValueCmd):

1763

command = command.SetChecker

1764

propname = "Checker"

1765

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

1766

1767

1768

class TestSetHostCmd(TestPropertyValueCmd):

1769

command = command.SetHost

1770

propname = "Host"

1771

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

1772

1773

1774

class TestSetSecretCmd(TestPropertyValueCmd):

1775

command = command.SetSecret

1776

propname = "Secret"

1777

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

1778

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

1779

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

1780

1781

1782

class TestSetTimeoutCmd(TestPropertyValueCmd):

1783

command = command.SetTimeout

1784

propname = "Timeout"

1785

values_to_set = [datetime.timedelta(),

1786

datetime.timedelta(minutes=5),

1787

datetime.timedelta(seconds=1),

1788

datetime.timedelta(weeks=1),

1789

datetime.timedelta(weeks=52)]

1790

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

1791

1792

1793

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1794

command = command.SetExtendedTimeout

1795

propname = "ExtendedTimeout"

1796

values_to_set = [datetime.timedelta(),

1797

datetime.timedelta(minutes=5),

1798

datetime.timedelta(seconds=1),

1799

datetime.timedelta(weeks=1),

1800

datetime.timedelta(weeks=52)]

1801

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

1802

1803

1804

class TestSetIntervalCmd(TestPropertyValueCmd):

1805

command = command.SetInterval

1806

propname = "Interval"

1807

values_to_set = [datetime.timedelta(),

1808

datetime.timedelta(minutes=5),

1809

datetime.timedelta(seconds=1),

1810

datetime.timedelta(weeks=1),

1811

datetime.timedelta(weeks=52)]

1812

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

1813

1814

1815

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1816

command = command.SetApprovalDelay

1817

propname = "ApprovalDelay"

1818

values_to_set = [datetime.timedelta(),

1819

datetime.timedelta(minutes=5),

1820

datetime.timedelta(seconds=1),

1821

datetime.timedelta(weeks=1),

1822

datetime.timedelta(weeks=52)]

1823

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

1824

1825

1826

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1827

command = command.SetApprovalDuration

1828

propname = "ApprovalDuration"

1829

values_to_set = [datetime.timedelta(),

1830

datetime.timedelta(minutes=5),

1831

datetime.timedelta(seconds=1),

1832

datetime.timedelta(weeks=1),

1833

datetime.timedelta(weeks=52)]

1834

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

1835

1836

1837

1838

def should_only_run_tests():

1839

parser = argparse.ArgumentParser(add_help=False)

1840

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

1841

args, unknown_args = parser.parse_known_args()

1842

run_tests = args.check

1843

if run_tests:

1844

# Remove --check argument from sys.argv

1845

sys.argv[1:] = unknown_args

1846

return run_tests

1847

1848

# Add all tests from doctest strings

1849

def load_tests(loader, tests, none):

1850

import doctest

1851

tests.addTests(doctest.DocTestSuite())

1852

return tests

354

1853

355

1854

if __name__ == "__main__":

356

main()

1855

try:

1856

if should_only_run_tests():

1857

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

1858

unittest.main()

1859

else:

1860

main()

1861

finally:

1862

logging.shutdown()