/mandos/trunk : revision 1086

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

64

import StringIO

65

io.StringIO = StringIO.StringIO

38

66

39

67

locale.setlocale(locale.LC_ALL, "")

40

68

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

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

69

dbus_busname_domain = "se.recompile"

70

dbus_busname = dbus_busname_domain + ".Mandos"

71

server_dbus_path = "/"

72

server_dbus_interface = dbus_busname_domain + ".Mandos"

73

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

74

del dbus_busname_domain

75

version = "1.8.3"

76

77

78

try:

79

dbus.OBJECT_MANAGER_IFACE

80

except AttributeError:

81

dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"

82

161

83

162

84

def main():

163

85

parser = argparse.ArgumentParser()

86

87

add_command_line_options(parser)

88

89

options = parser.parse_args()

90

91

check_option_syntax(parser, options)

92

93

clientnames = options.client

94

95

if options.debug:

96

log.setLevel(logging.DEBUG)

97

98

bus = dbus.SystemBus()

99

100

mandos_dbus_object = get_mandos_dbus_object(bus)

101

102

mandos_serv = dbus.Interface(

103

mandos_dbus_object, dbus_interface=server_dbus_interface)

104

mandos_serv_object_manager = dbus.Interface(

105

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

106

107

managed_objects = get_managed_objects(mandos_serv_object_manager)

108

109

all_clients = {}

110

for path, ifs_and_props in managed_objects.items():

111

try:

112

all_clients[path] = ifs_and_props[client_dbus_interface]

113

except KeyError:

114

pass

115

116

# Compile dict of (clientpath: properties) to process

117

if not clientnames:

118

clients = all_clients

119

else:

120

clients = {}

121

for name in clientnames:

122

for objpath, properties in all_clients.items():

123

if properties["Name"] == name:

124

clients[objpath] = properties

125

break

126

else:

127

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

128

sys.exit(1)

129

130

commands = commands_from_options(options)

131

132

for command in commands:

133

command.run(clients, bus, mandos_serv)

134

135

136

def add_command_line_options(parser):

164

137

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

165

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

138

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

166

139

help="show version number and exit")

167

140

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

168

141

help="Select all clients")

169

142

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

170

143

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

144

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

145

help="Dump client data in JSON format")

146

enable_disable = parser.add_mutually_exclusive_group()

147

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

148

help="Enable client")

149

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

150

action="store_true",

151

help="disable client")

175

152

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

176

153

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

154

start_stop_checker = parser.add_mutually_exclusive_group()

155

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

156

action="store_true",

157

help="Start checker for client")

158

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

159

action="store_true",

160

help="Stop checker for client")

181

161

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

182

162

help="Check if client is enabled")

183

163

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

184

164

help="Remove client")

185

165

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

186

166

help="Set checker command for client")

187

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

167

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

188

168

help="Set timeout for client")

189

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

169

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

190

170

help="Set extended timeout for client")

191

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

171

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

192

172

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

173

approve_deny_default = parser.add_mutually_exclusive_group()

174

approve_deny_default.add_argument(

175

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

176

default=None, dest="approved_by_default",

177

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

178

approve_deny_default.add_argument(

179

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

180

dest="approved_by_default",

181

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

182

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

200

183

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

201

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

184

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

202

185

help="Set duration of one client approval")

203

186

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

204

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

187

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

188

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

205

189

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

190

approve_deny = parser.add_mutually_exclusive_group()

191

approve_deny.add_argument(

192

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

193

help="Approve any current client request")

194

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

195

help="Deny any current client request")

196

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

197

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

198

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

199

help="Run self-test")

210

200

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

211

options = parser.parse_args()

212

201

202

203

def string_to_delta(interval):

204

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

205

206

try:

207

return rfc3339_duration_to_delta(interval)

208

except ValueError as e:

209

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

210

' '.join(e.args))

211

return parse_pre_1_6_1_interval(interval)

212

213

214

def rfc3339_duration_to_delta(duration):

215

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

216

217

>>> rfc3339_duration_to_delta("P7D")

218

datetime.timedelta(7)

219

>>> rfc3339_duration_to_delta("PT60S")

220

datetime.timedelta(0, 60)

221

>>> rfc3339_duration_to_delta("PT60M")

222

datetime.timedelta(0, 3600)

223

>>> rfc3339_duration_to_delta("P60M")

224

datetime.timedelta(1680)

225

>>> rfc3339_duration_to_delta("PT24H")

226

datetime.timedelta(1)

227

>>> rfc3339_duration_to_delta("P1W")

228

datetime.timedelta(7)

229

>>> rfc3339_duration_to_delta("PT5M30S")

230

datetime.timedelta(0, 330)

231

>>> rfc3339_duration_to_delta("P1DT3M20S")

232

datetime.timedelta(1, 200)

233

>>> # Can not be empty:

234

>>> rfc3339_duration_to_delta("")

235

Traceback (most recent call last):

236

...

237

ValueError: Invalid RFC 3339 duration: ""

238

>>> # Must start with "P":

239

>>> rfc3339_duration_to_delta("1D")

240

Traceback (most recent call last):

241

...

242

ValueError: Invalid RFC 3339 duration: "1D"

243

>>> # Must use correct order

244

>>> rfc3339_duration_to_delta("PT1S2M")

245

Traceback (most recent call last):

246

...

247

ValueError: Invalid RFC 3339 duration: "PT1S2M"

248

>>> # Time needs time marker

249

>>> rfc3339_duration_to_delta("P1H2S")

250

Traceback (most recent call last):

251

...

252

ValueError: Invalid RFC 3339 duration: "P1H2S"

253

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

254

>>> rfc3339_duration_to_delta("P1D2W")

255

Traceback (most recent call last):

256

...

257

ValueError: Invalid RFC 3339 duration: "P1D2W"

258

>>> rfc3339_duration_to_delta("P2W2H")

259

Traceback (most recent call last):

260

...

261

ValueError: Invalid RFC 3339 duration: "P2W2H"

262

"""

263

264

# Parsing an RFC 3339 duration with regular expressions is not

265

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

266

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

267

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

268

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

269

# avoid excessive use of external libraries.

270

271

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

272

Token = collections.namedtuple("Token", (

273

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

274

# a "group" containing digits

275

"value", # datetime.timedelta or None

276

"followers")) # Tokens valid after this token

277

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

278

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

279

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

280

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

281

datetime.timedelta(seconds=1),

282

frozenset((token_end, )))

283

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

284

datetime.timedelta(minutes=1),

285

frozenset((token_second, token_end)))

286

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

287

datetime.timedelta(hours=1),

288

frozenset((token_minute, token_end)))

289

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

290

None,

291

frozenset((token_hour, token_minute,

292

token_second)))

293

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

294

datetime.timedelta(days=1),

295

frozenset((token_time, token_end)))

296

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

297

datetime.timedelta(weeks=4),

298

frozenset((token_day, token_end)))

299

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

300

datetime.timedelta(weeks=52),

301

frozenset((token_month, token_end)))

302

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

303

datetime.timedelta(weeks=1),

304

frozenset((token_end, )))

305

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

306

frozenset((token_year, token_month,

307

token_day, token_time,

308

token_week)))

309

# Define starting values:

310

# Value so far

311

value = datetime.timedelta()

312

found_token = None

313

# Following valid tokens

314

followers = frozenset((token_duration, ))

315

# String left to parse

316

s = duration

317

# Loop until end token is found

318

while found_token is not token_end:

319

# Search for any currently valid tokens

320

for token in followers:

321

match = token.regexp.match(s)

322

if match is not None:

323

# Token found

324

if token.value is not None:

325

# Value found, parse digits

326

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

327

# Add to value so far

328

value += factor * token.value

329

# Strip token from string

330

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

331

# Go to found token

332

found_token = token

333

# Set valid next tokens

334

followers = found_token.followers

335

break

336

else:

337

# No currently valid tokens were found

338

raise ValueError("Invalid RFC 3339 duration: \"{}\""

339

.format(duration))

340

# End token found

341

return value

342

343

344

def parse_pre_1_6_1_interval(interval):

345

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

346

and return a datetime.timedelta

347

348

>>> parse_pre_1_6_1_interval('7d')

349

datetime.timedelta(7)

350

>>> parse_pre_1_6_1_interval('60s')

351

datetime.timedelta(0, 60)

352

>>> parse_pre_1_6_1_interval('60m')

353

datetime.timedelta(0, 3600)

354

>>> parse_pre_1_6_1_interval('24h')

355

datetime.timedelta(1)

356

>>> parse_pre_1_6_1_interval('1w')

357

datetime.timedelta(7)

358

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

359

datetime.timedelta(0, 330)

360

>>> parse_pre_1_6_1_interval('')

361

datetime.timedelta(0)

362

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

363

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

364

datetime.timedelta(2, 480, 18000)

365

366

"""

367

368

value = datetime.timedelta(0)

369

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

370

371

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

372

if suffix == "d":

373

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

374

elif suffix == "s":

375

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

376

elif suffix == "m":

377

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

378

elif suffix == "h":

379

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

380

elif suffix == "w":

381

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

382

elif suffix == "":

383

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

384

return value

385

386

387

def check_option_syntax(parser, options):

388

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

389

argparse"""

390

391

def has_actions(options):

392

return any((options.enable,

393

options.disable,

394

options.bump_timeout,

395

options.start_checker,

396

options.stop_checker,

397

options.is_enabled,

398

options.remove,

399

options.checker is not None,

400

options.timeout is not None,

401

options.extended_timeout is not None,

402

options.interval is not None,

403

options.approved_by_default is not None,

404

options.approval_delay is not None,

405

options.approval_duration is not None,

406

options.host is not None,

407

options.secret is not None,

408

options.approve,

409

options.deny))

410

213

411

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

214

412

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

215

413

if options.verbose and has_actions(options):

216

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

217

" --all.")

414

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

415

if options.dump_json and (options.verbose

416

or has_actions(options)):

417

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

218

418

if options.all and not has_actions(options):

219

419

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:

238

try:

239

mandos_clients = mandos_serv.GetAllClientsWithProperties()

240

finally:

241

#restore stderr

242

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

243

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

420

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

421

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

422

if options.remove:

423

options.remove = False

424

if has_actions(options) and not options.deny:

425

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

426

options.remove = True

427

428

429

def get_mandos_dbus_object(bus):

430

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

431

dbus_busname, server_dbus_path)

432

with if_dbus_exception_log_with_exception_and_exit(

433

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

434

mandos_dbus_object = bus.get_object(dbus_busname,

435

server_dbus_path)

436

return mandos_dbus_object

437

438

439

@contextlib.contextmanager

440

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

441

try:

442

yield

443

except dbus.exceptions.DBusException as e:

444

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

445

sys.exit(1)

446

447

448

def get_managed_objects(object_manager):

449

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

450

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

451

with if_dbus_exception_log_with_exception_and_exit(

452

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

453

with SilenceLogger("dbus.proxies"):

454

managed_objects = object_manager.GetManagedObjects()

455

return managed_objects

456

457

458

class SilenceLogger(object):

459

"Simple context manager to silence a particular logger"

460

def __init__(self, loggername):

461

self.logger = logging.getLogger(loggername)

462

463

def __enter__(self):

464

self.logger.addFilter(self.nullfilter)

465

return self

466

467

class NullFilter(logging.Filter):

468

def filter(self, record):

469

return False

470

471

nullfilter = NullFilter()

472

473

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

474

self.logger.removeFilter(self.nullfilter)

475

476

477

def commands_from_options(options):

478

479

commands = []

480

481

if options.is_enabled:

482

commands.append(command.IsEnabled())

483

484

if options.approve:

485

commands.append(command.Approve())

486

487

if options.deny:

488

commands.append(command.Deny())

489

490

if options.remove:

491

commands.append(command.Remove())

492

493

if options.dump_json:

494

commands.append(command.DumpJSON())

495

496

if options.enable:

497

commands.append(command.Enable())

498

499

if options.disable:

500

commands.append(command.Disable())

501

502

if options.bump_timeout:

503

commands.append(command.BumpTimeout())

504

505

if options.start_checker:

506

commands.append(command.StartChecker())

507

508

if options.stop_checker:

509

commands.append(command.StopChecker())

510

511

if options.approved_by_default is not None:

512

if options.approved_by_default:

513

commands.append(command.ApproveByDefault())

514

else:

515

commands.append(command.DenyByDefault())

516

517

if options.checker is not None:

518

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

519

520

if options.host is not None:

521

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

522

523

if options.secret is not None:

524

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

525

526

if options.timeout is not None:

527

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

528

529

if options.extended_timeout:

530

commands.append(

531

command.SetExtendedTimeout(options.extended_timeout))

532

533

if options.interval is not None:

534

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

535

536

if options.approval_delay is not None:

537

commands.append(

538

command.SetApprovalDelay(options.approval_delay))

539

540

if options.approval_duration is not None:

541

commands.append(

542

command.SetApprovalDuration(options.approval_duration))

543

544

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

545

# optionally verbosely

546

if not commands:

547

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

548

549

return commands

550

551

552

class command(object):

553

"""A namespace for command classes"""

554

555

class Base(object):

556

"""Abstract base class for commands"""

557

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

558

"""Normal commands should implement run_on_one_client(),

559

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

560

override this run() method instead.

561

"""

562

self.mandos = mandos

563

for clientpath, properties in clients.items():

564

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

565

dbus_busname, str(clientpath))

566

client = bus.get_object(dbus_busname, clientpath)

567

self.run_on_one_client(client, properties)

568

569

570

class IsEnabled(Base):

571

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

572

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

573

if self.is_enabled(client, properties):

574

sys.exit(0)

575

sys.exit(1)

576

def is_enabled(self, client, properties):

577

return properties["Enabled"]

578

579

580

class Approve(Base):

581

def run_on_one_client(self, client, properties):

582

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

583

client.__dbus_object_path__,

584

client_dbus_interface)

585

client.Approve(dbus.Boolean(True),

586

dbus_interface=client_dbus_interface)

587

588

589

class Deny(Base):

590

def run_on_one_client(self, client, properties):

591

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

592

client.__dbus_object_path__,

593

client_dbus_interface)

594

client.Approve(dbus.Boolean(False),

595

dbus_interface=client_dbus_interface)

596

597

598

class Remove(Base):

599

def run_on_one_client(self, client, properties):

600

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

601

dbus_busname, server_dbus_path,

602

server_dbus_interface,

603

str(client.__dbus_object_path__))

604

self.mandos.RemoveClient(client.__dbus_object_path__)

605

606

607

class Output(Base):

608

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

609

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

610

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

611

"Fingerprint", "CheckerRunning",

273

612

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

613

"ApprovedByDefault", "LastApprovalRequest",

614

"ApprovalDelay", "ApprovalDuration",

615

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

616

"LastCheckerStatus")

617

618

619

class DumpJSON(Output):

620

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

621

data = {client["Name"]:

622

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

623

for key in self.all_keywords}

624

for client in clients.values()}

625

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

626

627

@staticmethod

628

def dbus_boolean_to_bool(value):

629

if isinstance(value, dbus.Boolean):

630

value = bool(value)

631

return value

632

633

634

class PrintTable(Output):

635

def __init__(self, verbose=False):

636

self.verbose = verbose

637

638

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

639

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

640

"LastCheckedOK")

641

keywords = default_keywords

642

if self.verbose:

643

keywords = self.all_keywords

644

print(self.TableOfClients(clients.values(), keywords))

645

646

class TableOfClients(object):

647

tableheaders = {

648

"Name": "Name",

649

"Enabled": "Enabled",

650

"Timeout": "Timeout",

651

"LastCheckedOK": "Last Successful Check",

652

"LastApprovalRequest": "Last Approval Request",

653

"Created": "Created",

654

"Interval": "Interval",

655

"Host": "Host",

656

"Fingerprint": "Fingerprint",

657

"KeyID": "Key ID",

658

"CheckerRunning": "Check Is Running",

659

"LastEnabled": "Last Enabled",

660

"ApprovalPending": "Approval Is Pending",

661

"ApprovedByDefault": "Approved By Default",

662

"ApprovalDelay": "Approval Delay",

663

"ApprovalDuration": "Approval Duration",

664

"Checker": "Checker",

665

"ExtendedTimeout": "Extended Timeout",

666

"Expires": "Expires",

667

"LastCheckerStatus": "Last Checker Status",

668

}

669

670

def __init__(self, clients, keywords):

671

self.clients = clients

672

self.keywords = keywords

673

674

def __str__(self):

675

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

676

677

if sys.version_info.major == 2:

678

__unicode__ = __str__

679

def __str__(self):

680

return str(self).encode(

681

locale.getpreferredencoding())

682

683

def rows(self):

684

format_string = self.row_formatting_string()

685

rows = [self.header_line(format_string)]

686

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

687

for client in self.clients)

688

return rows

689

690

def row_formatting_string(self):

691

"Format string used to format table rows"

692

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

693

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

694

*(len(self.string_from_client(client,

695

key))

696

for client in self.clients)),

697

key=key)

698

for key in self.keywords)

699

700

def string_from_client(self, client, key):

701

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

702

703

@classmethod

704

def valuetostring(cls, value, keyword):

705

if isinstance(value, dbus.Boolean):

706

return "Yes" if value else "No"

707

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

708

"ApprovalDuration", "ExtendedTimeout"):

709

return cls.milliseconds_to_string(value)

710

return str(value)

711

712

def header_line(self, format_string):

713

return format_string.format(**self.tableheaders)

714

715

def client_line(self, client, format_string):

716

return format_string.format(

717

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

718

for key in self.keywords})

719

720

@staticmethod

721

def milliseconds_to_string(ms):

722

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

723

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

724

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

725

if td.days else "",

726

hours=td.seconds // 3600,

727

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

728

seconds=td.seconds % 60))

729

730

731

class Property(Base):

732

"Abstract class for Actions for setting one client property"

733

734

def run_on_one_client(self, client, properties):

735

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

736

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

737

client.__dbus_object_path__,

738

dbus.PROPERTIES_IFACE, client_dbus_interface,

739

self.propname, self.value_to_set

740

if not isinstance(self.value_to_set,

741

dbus.Boolean)

742

else bool(self.value_to_set))

743

client.Set(client_dbus_interface, self.propname,

744

self.value_to_set,

745

dbus_interface=dbus.PROPERTIES_IFACE)

746

747

@property

748

def propname(self):

749

raise NotImplementedError()

750

751

752

class Enable(Property):

753

propname = "Enabled"

754

value_to_set = dbus.Boolean(True)

755

756

757

class Disable(Property):

758

propname = "Enabled"

759

value_to_set = dbus.Boolean(False)

760

761

762

class BumpTimeout(Property):

763

propname = "LastCheckedOK"

764

value_to_set = ""

765

766

767

class StartChecker(Property):

768

propname = "CheckerRunning"

769

value_to_set = dbus.Boolean(True)

770

771

772

class StopChecker(Property):

773

propname = "CheckerRunning"

774

value_to_set = dbus.Boolean(False)

775

776

777

class ApproveByDefault(Property):

778

propname = "ApprovedByDefault"

779

value_to_set = dbus.Boolean(True)

780

781

782

class DenyByDefault(Property):

783

propname = "ApprovedByDefault"

784

value_to_set = dbus.Boolean(False)

785

786

787

class PropertyValue(Property):

788

"Abstract class for Property recieving a value as argument"

789

def __init__(self, value):

790

self.value_to_set = value

791

792

793

class SetChecker(PropertyValue):

794

propname = "Checker"

795

796

797

class SetHost(PropertyValue):

798

propname = "Host"

799

800

801

class SetSecret(PropertyValue):

802

propname = "Secret"

803

804

@property

805

def value_to_set(self):

806

return self._vts

807

808

@value_to_set.setter

809

def value_to_set(self, value):

810

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

811

self._vts = value.read()

812

value.close()

813

814

815

class MillisecondsPropertyValueArgument(PropertyValue):

816

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

817

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

818

819

@property

820

def value_to_set(self):

821

return self._vts

822

823

@value_to_set.setter

824

def value_to_set(self, value):

825

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

826

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

827

828

829

class SetTimeout(MillisecondsPropertyValueArgument):

830

propname = "Timeout"

831

832

833

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

834

propname = "ExtendedTimeout"

835

836

837

class SetInterval(MillisecondsPropertyValueArgument):

838

propname = "Interval"

839

840

841

class SetApprovalDelay(MillisecondsPropertyValueArgument):

842

propname = "ApprovalDelay"

843

844

845

class SetApprovalDuration(MillisecondsPropertyValueArgument):

846

propname = "ApprovalDuration"

847

848

849

850

class TestCaseWithAssertLogs(unittest.TestCase):

851

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

852

853

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

854

@contextlib.contextmanager

855

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

856

capturing_handler = self.CapturingLevelHandler(level)

857

old_level = logger.level

858

old_propagate = logger.propagate

859

logger.addHandler(capturing_handler)

860

logger.setLevel(level)

861

logger.propagate = False

862

try:

863

yield capturing_handler.watcher

864

finally:

865

logger.propagate = old_propagate

866

logger.removeHandler(capturing_handler)

867

logger.setLevel(old_level)

868

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

869

0)

870

871

class CapturingLevelHandler(logging.Handler):

872

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

873

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

874

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

875

def emit(self, record):

876

self.watcher.records.append(record)

877

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

878

879

LoggingWatcher = collections.namedtuple("LoggingWatcher",

880

("records",

881

"output"))

882

883

884

class Test_string_to_delta(TestCaseWithAssertLogs):

885

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

886

# rfc3339_duration_to_delta() already has more comprehensive

887

# tests, which is run by doctest.

888

889

def test_rfc3339_zero_seconds(self):

890

self.assertEqual(datetime.timedelta(),

891

string_to_delta("PT0S"))

892

893

def test_rfc3339_zero_days(self):

894

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

895

896

def test_rfc3339_one_second(self):

897

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

898

string_to_delta("PT1S"))

899

900

def test_rfc3339_two_hours(self):

901

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

902

string_to_delta("PT2H"))

903

904

def test_falls_back_to_pre_1_6_1_with_warning(self):

905

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

906

value = string_to_delta("2h")

907

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

908

909

910

class Test_check_option_syntax(unittest.TestCase):

911

def setUp(self):

912

self.parser = argparse.ArgumentParser()

913

add_command_line_options(self.parser)

914

915

def test_actions_requires_client_or_all(self):

916

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

917

options = self.parser.parse_args()

918

setattr(options, action, value)

919

with self.assertParseError():

920

self.check_option_syntax(options)

921

922

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

923

# check_option_syntax()

924

actions = {

925

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

926

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

927

# never used

928

"enable": True,

929

"disable": True,

930

"bump_timeout": True,

931

"start_checker": True,

932

"stop_checker": True,

933

"is_enabled": True,

934

"remove": True,

935

"checker": "x",

936

"timeout": datetime.timedelta(),

937

"extended_timeout": datetime.timedelta(),

938

"interval": datetime.timedelta(),

939

"approved_by_default": True,

940

"approval_delay": datetime.timedelta(),

941

"approval_duration": datetime.timedelta(),

942

"host": "x",

943

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

944

"approve": True,

945

"deny": True,

946

}

947

948

@contextlib.contextmanager

949

def assertParseError(self):

950

with self.assertRaises(SystemExit) as e:

951

with self.redirect_stderr_to_devnull():

952

yield

953

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

954

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

955

# /argparse.html#exiting-methods

956

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

957

958

@staticmethod

959

@contextlib.contextmanager

960

def redirect_stderr_to_devnull():

961

old_stderr = sys.stderr

962

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

963

sys.stderr = null

964

try:

965

yield

966

finally:

967

sys.stderr = old_stderr

968

969

def check_option_syntax(self, options):

970

check_option_syntax(self.parser, options)

971

972

def test_actions_all_conflicts_with_verbose(self):

973

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

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.all = True

977

options.verbose = True

978

with self.assertParseError():

979

self.check_option_syntax(options)

980

981

def test_actions_with_client_conflicts_with_verbose(self):

982

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

983

options = self.parser.parse_args()

984

setattr(options, action, value)

985

options.verbose = True

986

options.client = ["client"]

987

with self.assertParseError():

988

self.check_option_syntax(options)

989

990

def test_dump_json_conflicts_with_verbose(self):

991

options = self.parser.parse_args()

992

options.dump_json = True

993

options.verbose = True

994

with self.assertParseError():

995

self.check_option_syntax(options)

996

997

def test_dump_json_conflicts_with_action(self):

998

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

999

options = self.parser.parse_args()

1000

setattr(options, action, value)

1001

options.dump_json = True

1002

with self.assertParseError():

1003

self.check_option_syntax(options)

1004

1005

def test_all_can_not_be_alone(self):

1006

options = self.parser.parse_args()

1007

options.all = True

1008

with self.assertParseError():

1009

self.check_option_syntax(options)

1010

1011

def test_all_is_ok_with_any_action(self):

1012

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

1013

options = self.parser.parse_args()

1014

setattr(options, action, value)

1015

options.all = True

1016

self.check_option_syntax(options)

1017

1018

def test_any_action_is_ok_with_one_client(self):

1019

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

1020

options = self.parser.parse_args()

1021

setattr(options, action, value)

1022

options.client = ["client"]

1023

self.check_option_syntax(options)

1024

1025

def test_one_client_with_all_actions_except_is_enabled(self):

1026

options = self.parser.parse_args()

1027

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

1028

if action == "is_enabled":

1029

continue

1030

setattr(options, action, value)

1031

options.client = ["client"]

1032

self.check_option_syntax(options)

1033

1034

def test_two_clients_with_all_actions_except_is_enabled(self):

1035

options = self.parser.parse_args()

1036

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

1037

if action == "is_enabled":

1038

continue

1039

setattr(options, action, value)

1040

options.client = ["client1", "client2"]

1041

self.check_option_syntax(options)

1042

1043

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1044

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

1045

if action == "is_enabled":

1046

continue

1047

options = self.parser.parse_args()

1048

setattr(options, action, value)

1049

options.client = ["client1", "client2"]

1050

self.check_option_syntax(options)

1051

1052

def test_is_enabled_fails_without_client(self):

1053

options = self.parser.parse_args()

1054

options.is_enabled = True

1055

with self.assertParseError():

1056

self.check_option_syntax(options)

1057

1058

def test_is_enabled_fails_with_two_clients(self):

1059

options = self.parser.parse_args()

1060

options.is_enabled = True

1061

options.client = ["client1", "client2"]

1062

with self.assertParseError():

1063

self.check_option_syntax(options)

1064

1065

def test_remove_can_only_be_combined_with_action_deny(self):

1066

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

1067

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

1068

continue

1069

options = self.parser.parse_args()

1070

setattr(options, action, value)

1071

options.all = True

1072

options.remove = True

1073

with self.assertParseError():

1074

self.check_option_syntax(options)

1075

1076

1077

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1078

def test_calls_and_returns_get_object_on_bus(self):

1079

class MockBus(object):

1080

called = False

1081

def get_object(mockbus_self, busname, dbus_path):

1082

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

1083

# this MockBus instance is in "mockbus_self".

1084

self.assertEqual(dbus_busname, busname)

1085

self.assertEqual(server_dbus_path, dbus_path)

1086

mockbus_self.called = True

1087

return mockbus_self

1088

1089

mockbus = get_mandos_dbus_object(bus=MockBus())

1090

self.assertIsInstance(mockbus, MockBus)

1091

self.assertTrue(mockbus.called)

1092

1093

def test_logs_and_exits_on_dbus_error(self):

1094

class FailingBusStub(object):

1095

def get_object(self, busname, dbus_path):

1096

raise dbus.exceptions.DBusException("Test")

1097

1098

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

1099

with self.assertRaises(SystemExit) as e:

1100

bus = get_mandos_dbus_object(bus=FailingBusStub())

1101

1102

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

1103

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

1104

else:

1105

self.assertIsNotNone(e.exception.code)

1106

1107

1108

class Test_get_managed_objects(TestCaseWithAssertLogs):

1109

def test_calls_and_returns_GetManagedObjects(self):

1110

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

1111

class ObjectManagerStub(object):

1112

def GetManagedObjects(self):

1113

return managed_objects

1114

retval = get_managed_objects(ObjectManagerStub())

1115

self.assertDictEqual(managed_objects, retval)

1116

1117

def test_logs_and_exits_on_dbus_error(self):

1118

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

1119

1120

class ObjectManagerFailingStub(object):

1121

def GetManagedObjects(self):

1122

dbus_logger.error("Test")

1123

raise dbus.exceptions.DBusException("Test")

1124

1125

class CountingHandler(logging.Handler):

1126

count = 0

1127

def emit(self, record):

1128

self.count += 1

1129

1130

counting_handler = CountingHandler()

1131

1132

dbus_logger.addHandler(counting_handler)

1133

1134

try:

1135

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

1136

with self.assertRaises(SystemExit) as e:

1137

get_managed_objects(ObjectManagerFailingStub())

1138

finally:

1139

dbus_logger.removeFilter(counting_handler)

1140

1141

# Make sure the dbus logger was suppressed

1142

self.assertEqual(0, counting_handler.count)

1143

1144

# Test that the dbus_logger still works

1145

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

1146

dbus_logger.error("Test")

1147

1148

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

1149

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

1150

else:

1151

self.assertIsNotNone(e.exception.code)

1152

1153

1154

class Test_commands_from_options(unittest.TestCase):

1155

def setUp(self):

1156

self.parser = argparse.ArgumentParser()

1157

add_command_line_options(self.parser)

1158

1159

def test_is_enabled(self):

1160

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

1161

command.IsEnabled)

1162

1163

def assert_command_from_args(self, args, command_cls,

1164

**cmd_attrs):

1165

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

1166

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

1167

options = self.parser.parse_args(args)

1168

check_option_syntax(self.parser, options)

1169

commands = commands_from_options(options)

1170

self.assertEqual(1, len(commands))

1171

command = commands[0]

1172

self.assertIsInstance(command, command_cls)

1173

for key, value in cmd_attrs.items():

1174

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

1175

1176

def test_is_enabled_short(self):

1177

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

1178

command.IsEnabled)

1179

1180

def test_approve(self):

1181

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

1182

command.Approve)

1183

1184

def test_approve_short(self):

1185

self.assert_command_from_args(["-A", "client"],

1186

command.Approve)

1187

1188

def test_deny(self):

1189

self.assert_command_from_args(["--deny", "client"],

1190

command.Deny)

1191

1192

def test_deny_short(self):

1193

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

1194

1195

def test_remove(self):

1196

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

1197

command.Remove)

1198

1199

def test_deny_before_remove(self):

1200

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

1201

"client"])

1202

check_option_syntax(self.parser, options)

1203

commands = commands_from_options(options)

1204

self.assertEqual(2, len(commands))

1205

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

1206

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

1207

1208

def test_deny_before_remove_reversed(self):

1209

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

1210

"--all"])

1211

check_option_syntax(self.parser, options)

1212

commands = commands_from_options(options)

1213

self.assertEqual(2, len(commands))

1214

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

1215

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

1216

1217

def test_remove_short(self):

1218

self.assert_command_from_args(["-r", "client"],

1219

command.Remove)

1220

1221

def test_dump_json(self):

1222

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

1223

command.DumpJSON)

1224

1225

def test_enable(self):

1226

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

1227

command.Enable)

1228

1229

def test_enable_short(self):

1230

self.assert_command_from_args(["-e", "client"],

1231

command.Enable)

1232

1233

def test_disable(self):

1234

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

1235

command.Disable)

1236

1237

def test_disable_short(self):

1238

self.assert_command_from_args(["-d", "client"],

1239

command.Disable)

1240

1241

def test_bump_timeout(self):

1242

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

1243

command.BumpTimeout)

1244

1245

def test_bump_timeout_short(self):

1246

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

1247

command.BumpTimeout)

1248

1249

def test_start_checker(self):

1250

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

1251

command.StartChecker)

1252

1253

def test_stop_checker(self):

1254

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

1255

command.StopChecker)

1256

1257

def test_approve_by_default(self):

1258

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

1259

"client"],

1260

command.ApproveByDefault)

1261

1262

def test_deny_by_default(self):

1263

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

1264

command.DenyByDefault)

1265

1266

def test_checker(self):

1267

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

1268

command.SetChecker,

1269

value_to_set=":")

1270

1271

def test_checker_empty(self):

1272

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

1273

command.SetChecker,

1274

value_to_set="")

1275

1276

def test_checker_short(self):

1277

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

1278

command.SetChecker,

1279

value_to_set=":")

1280

1281

def test_host(self):

1282

self.assert_command_from_args(

1283

["--host", "client.example.org", "client"],

1284

command.SetHost, value_to_set="client.example.org")

1285

1286

def test_host_short(self):

1287

self.assert_command_from_args(

1288

["-H", "client.example.org", "client"], command.SetHost,

1289

value_to_set="client.example.org")

1290

1291

def test_secret_devnull(self):

1292

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

1293

"client"], command.SetSecret,

1294

value_to_set=b"")

1295

1296

def test_secret_tempfile(self):

1297

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

1298

value = b"secret\0xyzzy\nbar"

1299

f.write(value)

1300

f.seek(0)

1301

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

1302

"client"],

1303

command.SetSecret,

1304

value_to_set=value)

1305

1306

def test_secret_devnull_short(self):

1307

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

1308

"client"], command.SetSecret,

1309

value_to_set=b"")

1310

1311

def test_secret_tempfile_short(self):

1312

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

1313

value = b"secret\0xyzzy\nbar"

1314

f.write(value)

1315

f.seek(0)

1316

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

1317

command.SetSecret,

1318

value_to_set=value)

1319

1320

def test_timeout(self):

1321

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

1322

command.SetTimeout,

1323

value_to_set=300000)

1324

1325

def test_timeout_short(self):

1326

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

1327

command.SetTimeout,

1328

value_to_set=300000)

1329

1330

def test_extended_timeout(self):

1331

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

1332

"client"],

1333

command.SetExtendedTimeout,

1334

value_to_set=900000)

1335

1336

def test_interval(self):

1337

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

1338

"client"], command.SetInterval,

1339

value_to_set=120000)

1340

1341

def test_interval_short(self):

1342

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

1343

command.SetInterval,

1344

value_to_set=120000)

1345

1346

def test_approval_delay(self):

1347

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

1348

"client"],

1349

command.SetApprovalDelay,

1350

value_to_set=30000)

1351

1352

def test_approval_duration(self):

1353

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

1354

"client"],

1355

command.SetApprovalDuration,

1356

value_to_set=1000)

1357

1358

def test_print_table(self):

1359

self.assert_command_from_args([], command.PrintTable,

1360

verbose=False)

1361

1362

def test_print_table_verbose(self):

1363

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

1364

command.PrintTable,

1365

verbose=True)

1366

1367

def test_print_table_verbose_short(self):

1368

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

1369

verbose=True)

1370

1371

1372

class TestCommand(unittest.TestCase):

1373

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

1374

1375

def setUp(self):

1376

testcase = self

1377

class MockClient(object):

1378

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

1379

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

1380

self.attributes = attributes

1381

self.attributes["Name"] = name

1382

self.calls = []

1383

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

1384

testcase.assertEqual(client_dbus_interface, interface)

1385

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1386

dbus_interface)

1387

self.attributes[propname] = value

1388

def Approve(self, approve, dbus_interface):

1389

testcase.assertEqual(client_dbus_interface,

1390

dbus_interface)

1391

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

1392

dbus_interface)))

1393

self.client = MockClient(

1394

"foo",

1395

KeyID=("92ed150794387c03ce684574b1139a65"

1396

"94a34f895daaaf09fd8ea90a27cddb12"),

1397

Secret=b"secret",

1398

Host="foo.example.org",

1399

Enabled=dbus.Boolean(True),

1400

Timeout=300000,

1401

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

1402

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

1403

Interval=120000,

1404

Fingerprint=("778827225BA7DE539C5A"

1405

"7CFA59CFF7CDBD9A5920"),

1406

CheckerRunning=dbus.Boolean(False),

1407

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

1408

ApprovalPending=dbus.Boolean(False),

1409

ApprovedByDefault=dbus.Boolean(True),

1410

LastApprovalRequest="",

1411

ApprovalDelay=0,

1412

ApprovalDuration=1000,

1413

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

1414

ExtendedTimeout=900000,

1415

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

1416

LastCheckerStatus=0)

1417

self.other_client = MockClient(

1418

"barbar",

1419

KeyID=("0558568eedd67d622f5c83b35a115f79"

1420

"6ab612cff5ad227247e46c2b020f441c"),

1421

Secret=b"secretbar",

1422

Host="192.0.2.3",

1423

Enabled=dbus.Boolean(True),

1424

Timeout=300000,

1425

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

1426

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

1427

Interval=120000,

1428

Fingerprint=("3E393AEAEFB84C7E89E2"

1429

"F547B3A107558FCA3A27"),

1430

CheckerRunning=dbus.Boolean(True),

1431

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

1432

ApprovalPending=dbus.Boolean(False),

1433

ApprovedByDefault=dbus.Boolean(False),

1434

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

1435

ApprovalDelay=30000,

1436

ApprovalDuration=93785000,

1437

Checker=":",

1438

ExtendedTimeout=900000,

1439

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

1440

LastCheckerStatus=-2)

1441

self.clients = collections.OrderedDict(

1442

[

1443

(self.client.__dbus_object_path__,

1444

self.client.attributes),

1445

(self.other_client.__dbus_object_path__,

1446

self.other_client.attributes),

1447

])

1448

self.one_client = {self.client.__dbus_object_path__:

1449

self.client.attributes}

1450

1451

@property

1452

def bus(self):

1453

class MockBus(object):

1454

@staticmethod

1455

def get_object(client_bus_name, path):

1456

self.assertEqual(dbus_busname, client_bus_name)

1457

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

1458

# MockBus instance, since this is a static method!

1459

if path == self.client.__dbus_object_path__:

1460

return self.client

1461

elif path == self.other_client.__dbus_object_path__:

1462

return self.other_client

1463

return MockBus()

1464

1465

1466

class TestBaseCommands(TestCommand):

1467

1468

def test_IsEnabled_exits_successfully(self):

1469

with self.assertRaises(SystemExit) as e:

1470

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

1471

if e.exception.code is not None:

1472

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

1473

else:

1474

self.assertIsNone(e.exception.code)

1475

1476

def test_IsEnabled_exits_with_failure(self):

1477

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

1478

with self.assertRaises(SystemExit) as e:

1479

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

1480

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

1481

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

1482

else:

1483

self.assertIsNotNone(e.exception.code)

1484

1485

def test_Approve(self):

1486

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

1487

for clientpath in self.clients:

1488

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

1489

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

1490

client.calls)

1491

1492

def test_Deny(self):

1493

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

1494

for clientpath in self.clients:

1495

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

1496

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

1497

client.calls)

1498

1499

def test_Remove(self):

1500

class MandosSpy(object):

1501

def __init__(self):

1502

self.calls = []

1503

def RemoveClient(self, dbus_path):

1504

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

1505

mandos = MandosSpy()

1506

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

1507

for clientpath in self.clients:

1508

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

1509

mandos.calls)

1510

1511

expected_json = {

1512

"foo": {

1513

"Name": "foo",

1514

"KeyID": ("92ed150794387c03ce684574b1139a65"

1515

"94a34f895daaaf09fd8ea90a27cddb12"),

1516

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

1517

"Enabled": True,

1518

"Timeout": 300000,

1519

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

1520

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

1521

"Interval": 120000,

1522

"Fingerprint": ("778827225BA7DE539C5A"

1523

"7CFA59CFF7CDBD9A5920"),

1524

"CheckerRunning": False,

1525

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

1526

"ApprovalPending": False,

1527

"ApprovedByDefault": True,

1528

"LastApprovalRequest": "",

1529

"ApprovalDelay": 0,

1530

"ApprovalDuration": 1000,

1531

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

1532

"ExtendedTimeout": 900000,

1533

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

1534

"LastCheckerStatus": 0,

1535

},

1536

"barbar": {

1537

"Name": "barbar",

1538

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1539

"6ab612cff5ad227247e46c2b020f441c"),

1540

"Host": "192.0.2.3",

1541

"Enabled": True,

1542

"Timeout": 300000,

1543

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

1544

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

1545

"Interval": 120000,

1546

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1547

"F547B3A107558FCA3A27"),

1548

"CheckerRunning": True,

1549

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

1550

"ApprovalPending": False,

1551

"ApprovedByDefault": False,

1552

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

1553

"ApprovalDelay": 30000,

1554

"ApprovalDuration": 93785000,

1555

"Checker": ":",

1556

"ExtendedTimeout": 900000,

1557

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

1558

"LastCheckerStatus": -2,

1559

},

1560

}

1561

1562

def test_DumpJSON_normal(self):

1563

with self.capture_stdout_to_buffer() as buffer:

1564

command.DumpJSON().run(self.clients)

1565

json_data = json.loads(buffer.getvalue())

1566

self.assertDictEqual(self.expected_json, json_data)

1567

1568

@staticmethod

1569

@contextlib.contextmanager

1570

def capture_stdout_to_buffer():

1571

capture_buffer = io.StringIO()

1572

old_stdout = sys.stdout

1573

sys.stdout = capture_buffer

1574

try:

1575

yield capture_buffer

1576

finally:

1577

sys.stdout = old_stdout

1578

1579

def test_DumpJSON_one_client(self):

1580

with self.capture_stdout_to_buffer() as buffer:

1581

command.DumpJSON().run(self.one_client)

1582

json_data = json.loads(buffer.getvalue())

1583

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

1584

self.assertDictEqual(expected_json, json_data)

1585

1586

def test_PrintTable_normal(self):

1587

with self.capture_stdout_to_buffer() as buffer:

1588

command.PrintTable().run(self.clients)

1589

expected_output = "\n".join((

1590

"Name Enabled Timeout Last Successful Check",

1591

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

1592

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

1593

)) + "\n"

1594

self.assertEqual(expected_output, buffer.getvalue())

1595

1596

def test_PrintTable_verbose(self):

1597

with self.capture_stdout_to_buffer() as buffer:

1598

command.PrintTable(verbose=True).run(self.clients)

1599

columns = (

1600

(

1601

"Name ",

1602

"foo ",

1603

"barbar ",

1604

),(

1605

"Enabled ",

1606

"Yes ",

1607

"Yes ",

1608

),(

1609

"Timeout ",

1610

"00:05:00 ",

1611

"00:05:00 ",

1612

),(

1613

"Last Successful Check ",

1614

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

1615

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

1616

),(

1617

"Created ",

1618

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

1619

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

1620

),(

1621

"Interval ",

1622

"00:02:00 ",

1623

"00:02:00 ",

1624

),(

1625

"Host ",

1626

"foo.example.org ",

1627

"192.0.2.3 ",

1628

),(

1629

("Key ID "

1630

" "),

1631

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1632

"ea90a27cddb12 "),

1633

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1634

"46c2b020f441c "),

1635

),(

1636

"Fingerprint ",

1637

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1638

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1639

),(

1640

"Check Is Running ",

1641

"No ",

1642

"Yes ",

1643

),(

1644

"Last Enabled ",

1645

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

1646

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

1647

),(

1648

"Approval Is Pending ",

1649

"No ",

1650

"No ",

1651

),(

1652

"Approved By Default ",

1653

"Yes ",

1654

"No ",

1655

),(

1656

"Last Approval Request ",

1657

" ",

1658

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

1659

),(

1660

"Approval Delay ",

1661

"00:00:00 ",

1662

"00:00:30 ",

1663

),(

1664

"Approval Duration ",

1665

"00:00:01 ",

1666

"1T02:03:05 ",

1667

),(

1668

"Checker ",

1669

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

1670

": ",

1671

),(

1672

"Extended Timeout ",

1673

"00:15:00 ",

1674

"00:15:00 ",

1675

),(

1676

"Expires ",

1677

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

1678

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

1679

),(

1680

"Last Checker Status",

1681

"0 ",

1682

"-2 ",

1683

)

1684

)

1685

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

1686

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

1687

for rows in columns)

1688

for line in range(num_lines))

1689

+ "\n")

1690

self.assertEqual(expected_output, buffer.getvalue())

1691

1692

def test_PrintTable_one_client(self):

1693

with self.capture_stdout_to_buffer() as buffer:

1694

command.PrintTable().run(self.one_client)

1695

expected_output = "\n".join((

1696

"Name Enabled Timeout Last Successful Check",

1697

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

1698

)) + "\n"

1699

self.assertEqual(expected_output, buffer.getvalue())

1700

1701

1702

class TestPropertyCmd(TestCommand):

1703

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

1704

def runTest(self):

1705

if not hasattr(self, "command"):

1706

return

1707

values_to_get = getattr(self, "values_to_get",

1708

self.values_to_set)

1709

for value_to_set, value_to_get in zip(self.values_to_set,

1710

values_to_get):

1711

for clientpath in self.clients:

1712

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

1713

old_value = client.attributes[self.propname]

1714

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

1715

self.run_command(value_to_set, self.clients)

1716

for clientpath in self.clients:

1717

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

1718

value = client.attributes[self.propname]

1719

self.assertNotIsInstance(value, self.Unique)

1720

self.assertEqual(value_to_get, value)

1721

1722

class Unique(object):

1723

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

1724

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

1725

1726

def run_command(self, value, clients):

1727

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

1728

1729

1730

class TestEnableCmd(TestPropertyCmd):

1731

command = command.Enable

1732

propname = "Enabled"

1733

values_to_set = [dbus.Boolean(True)]

1734

1735

1736

class TestDisableCmd(TestPropertyCmd):

1737

command = command.Disable

1738

propname = "Enabled"

1739

values_to_set = [dbus.Boolean(False)]

1740

1741

1742

class TestBumpTimeoutCmd(TestPropertyCmd):

1743

command = command.BumpTimeout

1744

propname = "LastCheckedOK"

1745

values_to_set = [""]

1746

1747

1748

class TestStartCheckerCmd(TestPropertyCmd):

1749

command = command.StartChecker

1750

propname = "CheckerRunning"

1751

values_to_set = [dbus.Boolean(True)]

1752

1753

1754

class TestStopCheckerCmd(TestPropertyCmd):

1755

command = command.StopChecker

1756

propname = "CheckerRunning"

1757

values_to_set = [dbus.Boolean(False)]

1758

1759

1760

class TestApproveByDefaultCmd(TestPropertyCmd):

1761

command = command.ApproveByDefault

1762

propname = "ApprovedByDefault"

1763

values_to_set = [dbus.Boolean(True)]

1764

1765

1766

class TestDenyByDefaultCmd(TestPropertyCmd):

1767

command = command.DenyByDefault

1768

propname = "ApprovedByDefault"

1769

values_to_set = [dbus.Boolean(False)]

1770

1771

1772

class TestPropertyValueCmd(TestPropertyCmd):

1773

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

1774

1775

def runTest(self):

1776

if type(self) is TestPropertyValueCmd:

1777

return

1778

return super(TestPropertyValueCmd, self).runTest()

1779

1780

def run_command(self, value, clients):

1781

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

1782

1783

1784

class TestSetCheckerCmd(TestPropertyValueCmd):

1785

command = command.SetChecker

1786

propname = "Checker"

1787

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

1788

1789

1790

class TestSetHostCmd(TestPropertyValueCmd):

1791

command = command.SetHost

1792

propname = "Host"

1793

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

1794

1795

1796

class TestSetSecretCmd(TestPropertyValueCmd):

1797

command = command.SetSecret

1798

propname = "Secret"

1799

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

1800

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

1801

values_to_get = [f.getvalue() for f in values_to_set]

1802

1803

1804

class TestSetTimeoutCmd(TestPropertyValueCmd):

1805

command = command.SetTimeout

1806

propname = "Timeout"

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 = [dt.total_seconds()*1000 for dt in values_to_set]

1813

1814

1815

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1816

command = command.SetExtendedTimeout

1817

propname = "ExtendedTimeout"

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 = [dt.total_seconds()*1000 for dt in values_to_set]

1824

1825

1826

class TestSetIntervalCmd(TestPropertyValueCmd):

1827

command = command.SetInterval

1828

propname = "Interval"

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 = [dt.total_seconds()*1000 for dt in values_to_set]

1835

1836

1837

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1838

command = command.SetApprovalDelay

1839

propname = "ApprovalDelay"

1840

values_to_set = [datetime.timedelta(),

1841

datetime.timedelta(minutes=5),

1842

datetime.timedelta(seconds=1),

1843

datetime.timedelta(weeks=1),

1844

datetime.timedelta(weeks=52)]

1845

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1846

1847

1848

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1849

command = command.SetApprovalDuration

1850

propname = "ApprovalDuration"

1851

values_to_set = [datetime.timedelta(),

1852

datetime.timedelta(minutes=5),

1853

datetime.timedelta(seconds=1),

1854

datetime.timedelta(weeks=1),

1855

datetime.timedelta(weeks=52)]

1856

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1857

1858

1859

1860

def should_only_run_tests():

1861

parser = argparse.ArgumentParser(add_help=False)

1862

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

1863

args, unknown_args = parser.parse_known_args()

1864

run_tests = args.check

1865

if run_tests:

1866

# Remove --check argument from sys.argv

1867

sys.argv[1:] = unknown_args

1868

return run_tests

1869

1870

# Add all tests from doctest strings

1871

def load_tests(loader, tests, none):

1872

import doctest

1873

tests.addTests(doctest.DocTestSuite())

1874

return tests

343

1875

344

1876

if __name__ == "__main__":

345

main()

1877

try:

1878

if should_only_run_tests():

1879

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

1880

unittest.main()

1881

else:

1882

main()

1883

finally:

1884

logging.shutdown()