/mandos/release : revision 237.7.624

1

#!/usr/bin/python

2

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

3

#

2

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This program is free software: you can redistribute it and/or modify

10

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

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

# Mandos is free software: you can redistribute it and/or modify it

12

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

11

13

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

12

14

# (at your option) any later version.

13

15

#

14

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

15

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

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

16

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17

19

# GNU General Public License for more details.

18

#

20

#

19

21

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

20

# along with this program. If not, see

21

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

22

#

22

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

23

#

23

24

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

24

#

25

#

25

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

29

35

import sys

30

import dbus

31

36

import argparse

32

37

import locale

33

38

import datetime

34

39

import re

35

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

log = logging.getLogger(sys.argv[0])

57

logging.basicConfig(level="INFO", # Show info level messages

58

format="%(message)s") # Show basic log messages

59

60

logging.captureWarnings(True) # Show warnings via the logging system

61

62

if sys.version_info.major == 2:

63

str = unicode

36

64

37

65

locale.setlocale(locale.LC_ALL, "")

38

66

39

tablewords = {

40

"Name": "Name",

41

"Enabled": "Enabled",

42

"Timeout": "Timeout",

43

"LastCheckedOK": "Last Successful Check",

44

"LastApprovalRequest": "Last Approval Request",

45

"Created": "Created",

46

"Interval": "Interval",

47

"Host": "Host",

48

"Fingerprint": "Fingerprint",

49

"CheckerRunning": "Check Is Running",

50

"LastEnabled": "Last Enabled",

51

"ApprovalPending": "Approval Is Pending",

52

"ApprovedByDefault": "Approved By Default",

53

"ApprovalDelay": "Approval Delay",

54

"ApprovalDuration": "Approval Duration",

55

"Checker": "Checker",

56

"ExtendedTimeout" : "Extended Timeout"

57

}

58

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

59

domain = "se.recompile"

60

busname = domain + ".Mandos"

61

server_path = "/"

62

server_interface = domain + ".Mandos"

63

client_interface = domain + ".Mandos.Client"

64

version = "1.5.3"

65

66

def timedelta_to_milliseconds(td):

67

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

68

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

69

+ (td.seconds * 1000)

70

+ (td.microseconds // 1000))

71

72

def milliseconds_to_string(ms):

73

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

74

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

75

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

76

hours = td.seconds // 3600,

77

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

78

seconds = td.seconds % 60,

79

))

80

81

def string_to_delta(interval):

82

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

83

84

>>> string_to_delta("7d")

85

datetime.timedelta(7)

86

>>> string_to_delta("60s")

87

datetime.timedelta(0, 60)

88

>>> string_to_delta("60m")

89

datetime.timedelta(0, 3600)

90

>>> string_to_delta("24h")

91

datetime.timedelta(1)

92

>>> string_to_delta("1w")

93

datetime.timedelta(7)

94

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

95

datetime.timedelta(0, 330)

96

"""

97

value = datetime.timedelta(0)

98

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

99

100

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

101

if suffix == "d":

102

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

103

elif suffix == "s":

104

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

105

elif suffix == "m":

106

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

107

elif suffix == "h":

108

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

109

elif suffix == "w":

110

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

111

elif suffix == "":

112

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

113

return value

114

115

def print_clients(clients, keywords):

116

def valuetostring(value, keyword):

117

if type(value) is dbus.Boolean:

118

return "Yes" if value else "No"

119

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

120

"ApprovalDuration", "ExtendedTimeout"):

121

return milliseconds_to_string(value)

122

return unicode(value)

123

124

# Create format string to print table rows

125

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

126

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

127

max(len(valuetostring(client[key],

128

key))

129

for client in

130

clients)),

131

key = key) for key in keywords)

132

# Print header line

133

print(format_string.format(**tablewords))

134

for client in clients:

135

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

136

valuetostring(client[key],

137

key))

138

for key in keywords)))

139

140

def has_actions(options):

141

return any((options.enable,

142

options.disable,

143

options.bump_timeout,

144

options.start_checker,

145

options.stop_checker,

146

options.is_enabled,

147

options.remove,

148

options.checker is not None,

149

options.timeout is not None,

150

options.extended_timeout is not None,

151

options.interval is not None,

152

options.approved_by_default is not None,

153

options.approval_delay is not None,

154

options.approval_duration is not None,

155

options.host is not None,

156

options.secret is not None,

157

options.approve,

158

options.deny))

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

159

81

160

82

def main():

161

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

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

126

sys.exit(1)

127

128

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

162

135

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

163

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

136

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

164

137

help="show version number and exit")

165

138

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

166

139

help="Select all clients")

167

140

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

168

141

help="Print all fields")

169

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

170

help="Enable client")

171

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

172

help="disable client")

142

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

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

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

146

help="Enable client")

147

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

148

action="store_true",

149

help="disable client")

173

150

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

174

151

help="Bump timeout for client")

175

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

176

help="Start checker for client")

177

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

178

help="Stop checker for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

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

154

action="store_true",

155

help="Start checker for client")

156

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

157

action="store_true",

158

help="Stop checker for client")

179

159

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

180

160

help="Check if client is enabled")

181

161

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

182

162

help="Remove client")

183

163

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

184

164

help="Set checker command for client")

185

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

165

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

186

166

help="Set timeout for client")

187

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

167

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

188

168

help="Set extended timeout for client")

189

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

169

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

190

170

help="Set checker interval for client")

191

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

192

default=None, dest="approved_by_default",

193

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

194

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

195

dest="approved_by_default",

196

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

197

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

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

175

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

176

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

179

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

180

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

198

181

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

199

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

182

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

200

183

help="Set duration of one client approval")

201

184

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

202

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

185

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

186

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

203

187

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

204

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

205

help="Approve any current client request")

206

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

207

help="Deny any current client request")

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

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

191

help="Approve any current client request")

192

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

193

help="Deny any current client request")

194

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

195

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

196

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

197

help="Run self-test")

208

198

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

209

options = parser.parse_args()

210

199

200

201

def string_to_delta(interval):

202

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

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

213

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

214

215

>>> rfc3339_duration_to_delta("P7D")

216

datetime.timedelta(7)

217

>>> rfc3339_duration_to_delta("PT60S")

218

datetime.timedelta(0, 60)

219

>>> rfc3339_duration_to_delta("PT60M")

220

datetime.timedelta(0, 3600)

221

>>> rfc3339_duration_to_delta("P60M")

222

datetime.timedelta(1680)

223

>>> rfc3339_duration_to_delta("PT24H")

224

datetime.timedelta(1)

225

>>> rfc3339_duration_to_delta("P1W")

226

datetime.timedelta(7)

227

>>> rfc3339_duration_to_delta("PT5M30S")

228

datetime.timedelta(0, 330)

229

>>> rfc3339_duration_to_delta("P1DT3M20S")

230

datetime.timedelta(1, 200)

231

>>> # Can not be empty:

232

>>> rfc3339_duration_to_delta("")

233

Traceback (most recent call last):

234

...

235

ValueError: Invalid RFC 3339 duration: ""

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

211

409

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

212

410

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

213

411

if options.verbose and has_actions(options):

214

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

215

" --all.")

412

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

415

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

216

416

if options.all and not has_actions(options):

217

417

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

218

219

try:

220

bus = dbus.SystemBus()

221

mandos_dbus_objc = bus.get_object(busname, server_path)

222

except dbus.exceptions.DBusException:

223

print("Could not connect to Mandos server",

224

file=sys.stderr)

225

sys.exit(1)

226

227

mandos_serv = dbus.Interface(mandos_dbus_objc,

228

dbus_interface = server_interface)

229

230

#block stderr since dbus library prints to stderr

231

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

232

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

233

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

234

os.close(null)

235

try:

418

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

419

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

420

if options.remove:

421

options.remove = False

422

if has_actions(options) and not options.deny:

423

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

424

options.remove = True

425

426

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

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

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

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

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

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

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

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

472

self.logger.removeFilter(self.nullfilter)

473

474

475

def commands_from_options(options):

476

477

commands = []

478

479

if options.is_enabled:

480

commands.append(IsEnabledCmd())

481

482

if options.approve:

483

commands.append(ApproveCmd())

484

485

if options.deny:

486

commands.append(DenyCmd())

487

488

if options.remove:

489

commands.append(RemoveCmd())

490

491

if options.dump_json:

492

commands.append(DumpJSONCmd())

493

494

if options.enable:

495

commands.append(EnableCmd())

496

497

if options.disable:

498

commands.append(DisableCmd())

499

500

if options.bump_timeout:

501

commands.append(BumpTimeoutCmd())

502

503

if options.start_checker:

504

commands.append(StartCheckerCmd())

505

506

if options.stop_checker:

507

commands.append(StopCheckerCmd())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(ApproveByDefaultCmd())

512

else:

513

commands.append(DenyByDefaultCmd())

514

515

if options.checker is not None:

516

commands.append(SetCheckerCmd(options.checker))

517

518

if options.host is not None:

519

commands.append(SetHostCmd(options.host))

520

521

if options.secret is not None:

522

commands.append(SetSecretCmd(options.secret))

523

524

if options.timeout is not None:

525

commands.append(SetTimeoutCmd(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

SetExtendedTimeoutCmd(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(SetIntervalCmd(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(SetApprovalDelayCmd(options.approval_delay))

536

537

if options.approval_duration is not None:

538

commands.append(

539

SetApprovalDurationCmd(options.approval_duration))

540

541

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

542

# optionally verbosely

543

if not commands:

544

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

545

546

return commands

547

548

549

class Command(object):

550

"""Abstract class for commands"""

551

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

552

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

553

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

554

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

555

self.mandos = mandos

556

for clientpath, properties in clients.items():

557

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

558

dbus_busname, str(clientpath))

559

client = bus.get_object(dbus_busname, clientpath)

560

self.run_on_one_client(client, properties)

561

562

563

class IsEnabledCmd(Command):

564

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

565

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

566

if self.is_enabled(client, properties):

567

sys.exit(0)

568

sys.exit(1)

569

def is_enabled(self, client, properties):

570

return properties["Enabled"]

571

572

573

class ApproveCmd(Command):

574

def run_on_one_client(self, client, properties):

575

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

576

client.__dbus_object_path__, client_dbus_interface)

577

client.Approve(dbus.Boolean(True),

578

dbus_interface=client_dbus_interface)

579

580

581

class DenyCmd(Command):

582

def run_on_one_client(self, client, properties):

583

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

584

client.__dbus_object_path__, client_dbus_interface)

585

client.Approve(dbus.Boolean(False),

586

dbus_interface=client_dbus_interface)

587

588

589

class RemoveCmd(Command):

590

def run_on_one_client(self, client, properties):

591

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

592

server_dbus_path, server_dbus_interface,

593

str(client.__dbus_object_path__))

594

self.mandos.RemoveClient(client.__dbus_object_path__)

595

596

597

class OutputCmd(Command):

598

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

599

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

600

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

601

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

602

"ApprovalPending", "ApprovedByDefault",

603

"LastApprovalRequest", "ApprovalDelay",

604

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

605

"Expires", "LastCheckerStatus")

606

607

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

608

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

609

610

def output(self, clients):

611

raise NotImplementedError()

612

613

614

class DumpJSONCmd(OutputCmd):

615

def output(self, clients):

616

data = {client["Name"]:

617

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

618

for key in self.all_keywords}

619

for client in clients}

620

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

621

622

@staticmethod

623

def dbus_boolean_to_bool(value):

624

if isinstance(value, dbus.Boolean):

625

value = bool(value)

626

return value

627

628

629

class PrintTableCmd(OutputCmd):

630

def __init__(self, verbose=False):

631

self.verbose = verbose

632

633

def output(self, clients):

634

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

635

"LastCheckedOK")

636

keywords = default_keywords

637

if self.verbose:

638

keywords = self.all_keywords

639

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

640

641

class TableOfClients(object):

642

tableheaders = {

643

"Name": "Name",

644

"Enabled": "Enabled",

645

"Timeout": "Timeout",

646

"LastCheckedOK": "Last Successful Check",

647

"LastApprovalRequest": "Last Approval Request",

648

"Created": "Created",

649

"Interval": "Interval",

650

"Host": "Host",

651

"Fingerprint": "Fingerprint",

652

"KeyID": "Key ID",

653

"CheckerRunning": "Check Is Running",

654

"LastEnabled": "Last Enabled",

655

"ApprovalPending": "Approval Is Pending",

656

"ApprovedByDefault": "Approved By Default",

657

"ApprovalDelay": "Approval Delay",

658

"ApprovalDuration": "Approval Duration",

659

"Checker": "Checker",

660

"ExtendedTimeout": "Extended Timeout",

661

"Expires": "Expires",

662

"LastCheckerStatus": "Last Checker Status",

663

}

664

665

def __init__(self, clients, keywords):

666

self.clients = clients

667

self.keywords = keywords

668

669

def __str__(self):

670

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

671

672

if sys.version_info.major == 2:

673

__unicode__ = __str__

674

def __str__(self):

675

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

676

677

def rows(self):

678

format_string = self.row_formatting_string()

679

rows = [self.header_line(format_string)]

680

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

681

for client in self.clients)

682

return rows

683

684

def row_formatting_string(self):

685

"Format string used to format table rows"

686

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

687

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

688

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

689

for client in self.clients)),

690

key=key)

691

for key in self.keywords)

692

693

def string_from_client(self, client, key):

694

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

695

696

@classmethod

697

def valuetostring(cls, value, keyword):

698

if isinstance(value, dbus.Boolean):

699

return "Yes" if value else "No"

700

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

701

"ApprovalDuration", "ExtendedTimeout"):

702

return cls.milliseconds_to_string(value)

703

return str(value)

704

705

def header_line(self, format_string):

706

return format_string.format(**self.tableheaders)

707

708

def client_line(self, client, format_string):

709

return format_string.format(

710

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

711

for key in self.keywords})

712

713

@staticmethod

714

def milliseconds_to_string(ms):

715

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

716

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

717

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

718

if td.days else "",

719

hours=td.seconds // 3600,

720

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

721

seconds=td.seconds % 60))

722

723

724

class PropertyCmd(Command):

725

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

726

727

def run_on_one_client(self, client, properties):

728

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

729

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

730

client.__dbus_object_path__,

731

dbus.PROPERTIES_IFACE, client_dbus_interface,

732

self.propname, self.value_to_set

733

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

734

else bool(self.value_to_set))

735

client.Set(client_dbus_interface, self.propname,

736

self.value_to_set,

737

dbus_interface=dbus.PROPERTIES_IFACE)

738

739

@property

740

def propname(self):

741

raise NotImplementedError()

742

743

744

class EnableCmd(PropertyCmd):

745

propname = "Enabled"

746

value_to_set = dbus.Boolean(True)

747

748

749

class DisableCmd(PropertyCmd):

750

propname = "Enabled"

751

value_to_set = dbus.Boolean(False)

752

753

754

class BumpTimeoutCmd(PropertyCmd):

755

propname = "LastCheckedOK"

756

value_to_set = ""

757

758

759

class StartCheckerCmd(PropertyCmd):

760

propname = "CheckerRunning"

761

value_to_set = dbus.Boolean(True)

762

763

764

class StopCheckerCmd(PropertyCmd):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(False)

767

768

769

class ApproveByDefaultCmd(PropertyCmd):

770

propname = "ApprovedByDefault"

771

value_to_set = dbus.Boolean(True)

772

773

774

class DenyByDefaultCmd(PropertyCmd):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(False)

777

778

779

class PropertyValueCmd(PropertyCmd):

780

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

781

def __init__(self, value):

782

self.value_to_set = value

783

784

785

class SetCheckerCmd(PropertyValueCmd):

786

propname = "Checker"

787

788

789

class SetHostCmd(PropertyValueCmd):

790

propname = "Host"

791

792

793

class SetSecretCmd(PropertyValueCmd):

794

propname = "Secret"

795

796

@property

797

def value_to_set(self):

798

return self._vts

799

800

@value_to_set.setter

801

def value_to_set(self, value):

802

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

803

self._vts = value.read()

804

value.close()

805

806

807

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

808

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

809

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

810

811

@property

812

def value_to_set(self):

813

return self._vts

814

815

@value_to_set.setter

816

def value_to_set(self, value):

817

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

818

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

819

820

821

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

822

propname = "Timeout"

823

824

825

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

826

propname = "ExtendedTimeout"

827

828

829

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

830

propname = "Interval"

831

832

833

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

834

propname = "ApprovalDelay"

835

836

837

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

838

propname = "ApprovalDuration"

839

840

841

842

class TestCaseWithAssertLogs(unittest.TestCase):

843

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

844

845

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

846

@contextlib.contextmanager

847

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

848

capturing_handler = self.CapturingLevelHandler(level)

849

old_level = logger.level

850

old_propagate = logger.propagate

851

logger.addHandler(capturing_handler)

852

logger.setLevel(level)

853

logger.propagate = False

854

try:

855

yield capturing_handler.watcher

856

finally:

857

logger.propagate = old_propagate

858

logger.removeHandler(capturing_handler)

859

logger.setLevel(old_level)

860

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

861

0)

862

863

class CapturingLevelHandler(logging.Handler):

864

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

865

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

866

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

867

def emit(self, record):

868

self.watcher.records.append(record)

869

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

870

871

LoggingWatcher = collections.namedtuple("LoggingWatcher",

872

("records",

873

"output"))

874

875

876

class Test_string_to_delta(TestCaseWithAssertLogs):

877

def test_handles_basic_rfc3339(self):

878

self.assertEqual(string_to_delta("PT0S"),

879

datetime.timedelta())

880

self.assertEqual(string_to_delta("P0D"),

881

datetime.timedelta())

882

self.assertEqual(string_to_delta("PT1S"),

883

datetime.timedelta(0, 1))

884

self.assertEqual(string_to_delta("PT2H"),

885

datetime.timedelta(0, 7200))

886

887

def test_falls_back_to_pre_1_6_1_with_warning(self):

888

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

889

value = string_to_delta("2h")

890

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

891

892

893

class Test_check_option_syntax(unittest.TestCase):

894

def setUp(self):

895

self.parser = argparse.ArgumentParser()

896

add_command_line_options(self.parser)

897

898

def test_actions_requires_client_or_all(self):

899

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

900

options = self.parser.parse_args()

901

setattr(options, action, value)

902

with self.assertParseError():

903

self.check_option_syntax(options)

904

905

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

906

# check_option_syntax()

907

actions = {

908

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

909

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

910

# never used

911

"enable": True,

912

"disable": True,

913

"bump_timeout": True,

914

"start_checker": True,

915

"stop_checker": True,

916

"is_enabled": True,

917

"remove": True,

918

"checker": "x",

919

"timeout": datetime.timedelta(),

920

"extended_timeout": datetime.timedelta(),

921

"interval": datetime.timedelta(),

922

"approved_by_default": True,

923

"approval_delay": datetime.timedelta(),

924

"approval_duration": datetime.timedelta(),

925

"host": "x",

926

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

927

"approve": True,

928

"deny": True,

929

}

930

931

@contextlib.contextmanager

932

def assertParseError(self):

933

with self.assertRaises(SystemExit) as e:

934

with self.redirect_stderr_to_devnull():

935

yield

936

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

937

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

938

# /argparse.html#exiting-methods

939

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

940

941

@staticmethod

942

@contextlib.contextmanager

943

def redirect_stderr_to_devnull():

944

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

945

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

946

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

947

os.close(null)

236

948

try:

237

mandos_clients = mandos_serv.GetAllClientsWithProperties()

949

yield

238

950

finally:

239

#restore stderr

951

# restore stderr

240

952

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

241

953

os.close(stderrcopy)

242

except dbus.exceptions.DBusException:

243

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

244

file=sys.stderr)

245

sys.exit(1)

246

247

# Compile dict of (clients: properties) to process

248

clients={}

249

250

if options.all or not options.client:

251

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

252

for path, properties in

253

mandos_clients.iteritems())

254

else:

255

for name in options.client:

256

for path, client in mandos_clients.iteritems():

257

if client["Name"] == name:

258

client_objc = bus.get_object(busname, path)

259

clients[client_objc] = client

260

break

261

else:

262

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

263

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

264

sys.exit(1)

265

266

if not has_actions(options) and clients:

267

if options.verbose:

268

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

269

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

270

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

271

"LastEnabled", "ApprovalPending",

272

"ApprovedByDefault",

273

"LastApprovalRequest", "ApprovalDelay",

274

"ApprovalDuration", "Checker",

275

"ExtendedTimeout")

276

else:

277

keywords = defaultkeywords

278

279

print_clients(clients.values(), keywords)

280

else:

281

# Process each client in the list by all selected options

282

for client in clients:

283

def set_client_prop(prop, value):

284

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

285

client.Set(client_interface, prop, value,

286

dbus_interface=dbus.PROPERTIES_IFACE)

287

def set_client_prop_ms(prop, value):

288

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

289

from a string to milliseconds."""

290

set_client_prop(prop,

291

timedelta_to_milliseconds

292

(string_to_delta(value)))

293

if options.remove:

294

mandos_serv.RemoveClient(client.__dbus_object_path__)

295

if options.enable:

296

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

297

if options.disable:

298

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

299

if options.bump_timeout:

300

set_client_prop("LastCheckedOK", "")

301

if options.start_checker:

302

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

303

if options.stop_checker:

304

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

305

if options.is_enabled:

306

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

307

"Enabled",

308

dbus_interface=

309

dbus.PROPERTIES_IFACE)

310

else 1)

311

if options.checker is not None:

312

set_client_prop("Checker", options.checker)

313

if options.host is not None:

314

set_client_prop("Host", options.host)

315

if options.interval is not None:

316

set_client_prop_ms("Interval", options.interval)

317

if options.approval_delay is not None:

318

set_client_prop_ms("ApprovalDelay",

319

options.approval_delay)

320

if options.approval_duration is not None:

321

set_client_prop_ms("ApprovalDuration",

322

options.approval_duration)

323

if options.timeout is not None:

324

set_client_prop_ms("Timeout", options.timeout)

325

if options.extended_timeout is not None:

326

set_client_prop_ms("ExtendedTimeout",

327

options.extended_timeout)

328

if options.secret is not None:

329

set_client_prop("Secret",

330

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

331

if options.approved_by_default is not None:

332

set_client_prop("ApprovedByDefault",

333

dbus.Boolean(options

334

.approved_by_default))

335

if options.approve:

336

client.Approve(dbus.Boolean(True),

337

dbus_interface=client_interface)

338

elif options.deny:

339

client.Approve(dbus.Boolean(False),

340

dbus_interface=client_interface)

954

955

def check_option_syntax(self, options):

956

check_option_syntax(self.parser, options)

957

958

def test_actions_all_conflicts_with_verbose(self):

959

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

960

options = self.parser.parse_args()

961

setattr(options, action, value)

962

options.all = True

963

options.verbose = True

964

with self.assertParseError():

965

self.check_option_syntax(options)

966

967

def test_actions_with_client_conflicts_with_verbose(self):

968

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

969

options = self.parser.parse_args()

970

setattr(options, action, value)

971

options.verbose = True

972

options.client = ["foo"]

973

with self.assertParseError():

974

self.check_option_syntax(options)

975

976

def test_dump_json_conflicts_with_verbose(self):

977

options = self.parser.parse_args()

978

options.dump_json = True

979

options.verbose = True

980

with self.assertParseError():

981

self.check_option_syntax(options)

982

983

def test_dump_json_conflicts_with_action(self):

984

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

985

options = self.parser.parse_args()

986

setattr(options, action, value)

987

options.dump_json = True

988

with self.assertParseError():

989

self.check_option_syntax(options)

990

991

def test_all_can_not_be_alone(self):

992

options = self.parser.parse_args()

993

options.all = True

994

with self.assertParseError():

995

self.check_option_syntax(options)

996

997

def test_all_is_ok_with_any_action(self):

998

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

999

options = self.parser.parse_args()

1000

setattr(options, action, value)

1001

options.all = True

1002

self.check_option_syntax(options)

1003

1004

def test_any_action_is_ok_with_one_client(self):

1005

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

1006

options = self.parser.parse_args()

1007

setattr(options, action, value)

1008

options.client = ["foo"]

1009

self.check_option_syntax(options)

1010

1011

def test_one_client_with_all_actions_except_is_enabled(self):

1012

options = self.parser.parse_args()

1013

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

1014

if action == "is_enabled":

1015

continue

1016

setattr(options, action, value)

1017

options.client = ["foo"]

1018

self.check_option_syntax(options)

1019

1020

def test_two_clients_with_all_actions_except_is_enabled(self):

1021

options = self.parser.parse_args()

1022

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

1023

if action == "is_enabled":

1024

continue

1025

setattr(options, action, value)

1026

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

1027

self.check_option_syntax(options)

1028

1029

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1030

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

1031

if action == "is_enabled":

1032

continue

1033

options = self.parser.parse_args()

1034

setattr(options, action, value)

1035

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

1036

self.check_option_syntax(options)

1037

1038

def test_is_enabled_fails_without_client(self):

1039

options = self.parser.parse_args()

1040

options.is_enabled = True

1041

with self.assertParseError():

1042

self.check_option_syntax(options)

1043

1044

def test_is_enabled_fails_with_two_clients(self):

1045

options = self.parser.parse_args()

1046

options.is_enabled = True

1047

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

1048

with self.assertParseError():

1049

self.check_option_syntax(options)

1050

1051

def test_remove_can_only_be_combined_with_action_deny(self):

1052

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

1053

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

1054

continue

1055

options = self.parser.parse_args()

1056

setattr(options, action, value)

1057

options.all = True

1058

options.remove = True

1059

with self.assertParseError():

1060

self.check_option_syntax(options)

1061

1062

1063

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1064

def test_calls_and_returns_get_object_on_bus(self):

1065

class MockBus(object):

1066

called = False

1067

def get_object(mockbus_self, busname, dbus_path):

1068

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

1069

# this MockBus instance is in "mockbus_self".

1070

self.assertEqual(busname, dbus_busname)

1071

self.assertEqual(dbus_path, server_dbus_path)

1072

mockbus_self.called = True

1073

return mockbus_self

1074

1075

mockbus = get_mandos_dbus_object(bus=MockBus())

1076

self.assertIsInstance(mockbus, MockBus)

1077

self.assertTrue(mockbus.called)

1078

1079

def test_logs_and_exits_on_dbus_error(self):

1080

class MockBusFailing(object):

1081

def get_object(self, busname, dbus_path):

1082

raise dbus.exceptions.DBusException("Test")

1083

1084

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

1085

with self.assertRaises(SystemExit) as e:

1086

bus = get_mandos_dbus_object(bus=MockBusFailing())

1087

1088

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

1089

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

1090

else:

1091

self.assertIsNotNone(e.exception.code)

1092

1093

1094

class Test_get_managed_objects(TestCaseWithAssertLogs):

1095

def test_calls_and_returns_GetManagedObjects(self):

1096

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

1097

class MockObjectManager(object):

1098

def GetManagedObjects(self):

1099

return managed_objects

1100

retval = get_managed_objects(MockObjectManager())

1101

self.assertDictEqual(managed_objects, retval)

1102

1103

def test_logs_and_exits_on_dbus_error(self):

1104

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

1105

1106

class MockObjectManagerFailing(object):

1107

def GetManagedObjects(self):

1108

dbus_logger.error("Test")

1109

raise dbus.exceptions.DBusException("Test")

1110

1111

class CountingHandler(logging.Handler):

1112

count = 0

1113

def emit(self, record):

1114

self.count += 1

1115

1116

counting_handler = CountingHandler()

1117

1118

dbus_logger.addHandler(counting_handler)

1119

1120

try:

1121

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

1122

with self.assertRaises(SystemExit) as e:

1123

get_managed_objects(MockObjectManagerFailing())

1124

finally:

1125

dbus_logger.removeFilter(counting_handler)

1126

1127

# Make sure the dbus logger was suppressed

1128

self.assertEqual(counting_handler.count, 0)

1129

1130

# Test that the dbus_logger still works

1131

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

1132

dbus_logger.error("Test")

1133

1134

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

1135

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

1136

else:

1137

self.assertIsNotNone(e.exception.code)

1138

1139

1140

class Test_commands_from_options(unittest.TestCase):

1141

def setUp(self):

1142

self.parser = argparse.ArgumentParser()

1143

add_command_line_options(self.parser)

1144

1145

def test_is_enabled(self):

1146

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

1147

IsEnabledCmd)

1148

1149

def assert_command_from_args(self, args, command_cls,

1150

**cmd_attrs):

1151

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

1152

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

1153

options = self.parser.parse_args(args)

1154

check_option_syntax(self.parser, options)

1155

commands = commands_from_options(options)

1156

self.assertEqual(len(commands), 1)

1157

command = commands[0]

1158

self.assertIsInstance(command, command_cls)

1159

for key, value in cmd_attrs.items():

1160

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

1161

1162

def test_is_enabled_short(self):

1163

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

1164

1165

def test_approve(self):

1166

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

1167

ApproveCmd)

1168

1169

def test_approve_short(self):

1170

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

1171

1172

def test_deny(self):

1173

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

1174

1175

def test_deny_short(self):

1176

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

1177

1178

def test_remove(self):

1179

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

1180

RemoveCmd)

1181

1182

def test_deny_before_remove(self):

1183

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

1184

"foo"])

1185

check_option_syntax(self.parser, options)

1186

commands = commands_from_options(options)

1187

self.assertEqual(len(commands), 2)

1188

self.assertIsInstance(commands[0], DenyCmd)

1189

self.assertIsInstance(commands[1], RemoveCmd)

1190

1191

def test_deny_before_remove_reversed(self):

1192

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

1193

"--all"])

1194

check_option_syntax(self.parser, options)

1195

commands = commands_from_options(options)

1196

self.assertEqual(len(commands), 2)

1197

self.assertIsInstance(commands[0], DenyCmd)

1198

self.assertIsInstance(commands[1], RemoveCmd)

1199

1200

def test_remove_short(self):

1201

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

1202

1203

def test_dump_json(self):

1204

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

1205

1206

def test_enable(self):

1207

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

1208

1209

def test_enable_short(self):

1210

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

1211

1212

def test_disable(self):

1213

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

1214

DisableCmd)

1215

1216

def test_disable_short(self):

1217

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

1218

1219

def test_bump_timeout(self):

1220

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

1221

BumpTimeoutCmd)

1222

1223

def test_bump_timeout_short(self):

1224

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

1225

1226

def test_start_checker(self):

1227

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

1228

StartCheckerCmd)

1229

1230

def test_stop_checker(self):

1231

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

1232

StopCheckerCmd)

1233

1234

def test_approve_by_default(self):

1235

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

1236

ApproveByDefaultCmd)

1237

1238

def test_deny_by_default(self):

1239

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

1240

DenyByDefaultCmd)

1241

1242

def test_checker(self):

1243

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

1244

SetCheckerCmd, value_to_set=":")

1245

1246

def test_checker_empty(self):

1247

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

1248

SetCheckerCmd, value_to_set="")

1249

1250

def test_checker_short(self):

1251

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

1252

SetCheckerCmd, value_to_set=":")

1253

1254

def test_host(self):

1255

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

1256

"foo"], SetHostCmd,

1257

value_to_set="foo.example.org")

1258

1259

def test_host_short(self):

1260

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

1261

"foo"], SetHostCmd,

1262

value_to_set="foo.example.org")

1263

1264

def test_secret_devnull(self):

1265

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

1266

"foo"], SetSecretCmd,

1267

value_to_set=b"")

1268

1269

def test_secret_tempfile(self):

1270

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

1271

value = b"secret\0xyzzy\nbar"

1272

f.write(value)

1273

f.seek(0)

1274

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

1275

"foo"], SetSecretCmd,

1276

value_to_set=value)

1277

1278

def test_secret_devnull_short(self):

1279

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

1280

SetSecretCmd, value_to_set=b"")

1281

1282

def test_secret_tempfile_short(self):

1283

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

1284

value = b"secret\0xyzzy\nbar"

1285

f.write(value)

1286

f.seek(0)

1287

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

1288

SetSecretCmd,

1289

value_to_set=value)

1290

1291

def test_timeout(self):

1292

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

1293

SetTimeoutCmd,

1294

value_to_set=300000)

1295

1296

def test_timeout_short(self):

1297

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

1298

SetTimeoutCmd,

1299

value_to_set=300000)

1300

1301

def test_extended_timeout(self):

1302

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

1303

"foo"],

1304

SetExtendedTimeoutCmd,

1305

value_to_set=900000)

1306

1307

def test_interval(self):

1308

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

1309

SetIntervalCmd,

1310

value_to_set=120000)

1311

1312

def test_interval_short(self):

1313

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

1314

SetIntervalCmd,

1315

value_to_set=120000)

1316

1317

def test_approval_delay(self):

1318

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

1319

"foo"], SetApprovalDelayCmd,

1320

value_to_set=30000)

1321

1322

def test_approval_duration(self):

1323

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

1324

"foo"], SetApprovalDurationCmd,

1325

value_to_set=1000)

1326

1327

def test_print_table(self):

1328

self.assert_command_from_args([], PrintTableCmd,

1329

verbose=False)

1330

1331

def test_print_table_verbose(self):

1332

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

1333

verbose=True)

1334

1335

def test_print_table_verbose_short(self):

1336

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

1337

verbose=True)

1338

1339

1340

class TestCmd(unittest.TestCase):

1341

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

1342

1343

def setUp(self):

1344

testcase = self

1345

class MockClient(object):

1346

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

1347

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

1348

self.attributes = attributes

1349

self.attributes["Name"] = name

1350

self.calls = []

1351

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

1352

testcase.assertEqual(interface, client_dbus_interface)

1353

testcase.assertEqual(dbus_interface,

1354

dbus.PROPERTIES_IFACE)

1355

self.attributes[propname] = value

1356

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

1357

testcase.assertEqual(interface, client_dbus_interface)

1358

testcase.assertEqual(dbus_interface,

1359

dbus.PROPERTIES_IFACE)

1360

return self.attributes[propname]

1361

def Approve(self, approve, dbus_interface):

1362

testcase.assertEqual(dbus_interface,

1363

client_dbus_interface)

1364

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

1365

dbus_interface)))

1366

self.client = MockClient(

1367

"foo",

1368

KeyID=("92ed150794387c03ce684574b1139a65"

1369

"94a34f895daaaf09fd8ea90a27cddb12"),

1370

Secret=b"secret",

1371

Host="foo.example.org",

1372

Enabled=dbus.Boolean(True),

1373

Timeout=300000,

1374

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

1375

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

1376

Interval=120000,

1377

Fingerprint=("778827225BA7DE539C5A"

1378

"7CFA59CFF7CDBD9A5920"),

1379

CheckerRunning=dbus.Boolean(False),

1380

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

1381

ApprovalPending=dbus.Boolean(False),

1382

ApprovedByDefault=dbus.Boolean(True),

1383

LastApprovalRequest="",

1384

ApprovalDelay=0,

1385

ApprovalDuration=1000,

1386

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

1387

ExtendedTimeout=900000,

1388

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

1389

LastCheckerStatus=0)

1390

self.other_client = MockClient(

1391

"barbar",

1392

KeyID=("0558568eedd67d622f5c83b35a115f79"

1393

"6ab612cff5ad227247e46c2b020f441c"),

1394

Secret=b"secretbar",

1395

Host="192.0.2.3",

1396

Enabled=dbus.Boolean(True),

1397

Timeout=300000,

1398

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

1399

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

1400

Interval=120000,

1401

Fingerprint=("3E393AEAEFB84C7E89E2"

1402

"F547B3A107558FCA3A27"),

1403

CheckerRunning=dbus.Boolean(True),

1404

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

1405

ApprovalPending=dbus.Boolean(False),

1406

ApprovedByDefault=dbus.Boolean(False),

1407

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

1408

ApprovalDelay=30000,

1409

ApprovalDuration=93785000,

1410

Checker=":",

1411

ExtendedTimeout=900000,

1412

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

1413

LastCheckerStatus=-2)

1414

self.clients = collections.OrderedDict(

1415

[

1416

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

1417

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

1418

])

1419

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

1420

1421

@property

1422

def bus(self):

1423

class Bus(object):

1424

@staticmethod

1425

def get_object(client_bus_name, path):

1426

self.assertEqual(client_bus_name, dbus_busname)

1427

return {

1428

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

1429

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

1430

"/clients/foo": self.client,

1431

"/clients/barbar": self.other_client,

1432

}[path]

1433

return Bus()

1434

1435

1436

class TestIsEnabledCmd(TestCmd):

1437

def test_is_enabled(self):

1438

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

1439

properties)

1440

for client, properties

1441

in self.clients.items()))

1442

1443

def test_is_enabled_run_exits_successfully(self):

1444

with self.assertRaises(SystemExit) as e:

1445

IsEnabledCmd().run(self.one_client)

1446

if e.exception.code is not None:

1447

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

1448

else:

1449

self.assertIsNone(e.exception.code)

1450

1451

def test_is_enabled_run_exits_with_failure(self):

1452

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

1453

with self.assertRaises(SystemExit) as e:

1454

IsEnabledCmd().run(self.one_client)

1455

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

1456

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

1457

else:

1458

self.assertIsNotNone(e.exception.code)

1459

1460

1461

class TestApproveCmd(TestCmd):

1462

def test_approve(self):

1463

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

1464

for clientpath in self.clients:

1465

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

1466

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

1467

client.calls)

1468

1469

1470

class TestDenyCmd(TestCmd):

1471

def test_deny(self):

1472

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

1473

for clientpath in self.clients:

1474

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

1475

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

1476

client.calls)

1477

1478

1479

class TestRemoveCmd(TestCmd):

1480

def test_remove(self):

1481

class MockMandos(object):

1482

def __init__(self):

1483

self.calls = []

1484

def RemoveClient(self, dbus_path):

1485

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

1486

mandos = MockMandos()

1487

super(TestRemoveCmd, self).setUp()

1488

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

1489

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

1490

for clientpath in self.clients:

1491

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

1492

mandos.calls)

1493

1494

1495

class TestDumpJSONCmd(TestCmd):

1496

def setUp(self):

1497

self.expected_json = {

1498

"foo": {

1499

"Name": "foo",

1500

"KeyID": ("92ed150794387c03ce684574b1139a65"

1501

"94a34f895daaaf09fd8ea90a27cddb12"),

1502

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

1503

"Enabled": True,

1504

"Timeout": 300000,

1505

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

1506

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

1507

"Interval": 120000,

1508

"Fingerprint": ("778827225BA7DE539C5A"

1509

"7CFA59CFF7CDBD9A5920"),

1510

"CheckerRunning": False,

1511

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

1512

"ApprovalPending": False,

1513

"ApprovedByDefault": True,

1514

"LastApprovalRequest": "",

1515

"ApprovalDelay": 0,

1516

"ApprovalDuration": 1000,

1517

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

1518

"ExtendedTimeout": 900000,

1519

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

1520

"LastCheckerStatus": 0,

1521

},

1522

"barbar": {

1523

"Name": "barbar",

1524

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1525

"6ab612cff5ad227247e46c2b020f441c"),

1526

"Host": "192.0.2.3",

1527

"Enabled": True,

1528

"Timeout": 300000,

1529

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

1530

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

1531

"Interval": 120000,

1532

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1533

"F547B3A107558FCA3A27"),

1534

"CheckerRunning": True,

1535

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

1536

"ApprovalPending": False,

1537

"ApprovedByDefault": False,

1538

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

1539

"ApprovalDelay": 30000,

1540

"ApprovalDuration": 93785000,

1541

"Checker": ":",

1542

"ExtendedTimeout": 900000,

1543

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

1544

"LastCheckerStatus": -2,

1545

},

1546

}

1547

return super(TestDumpJSONCmd, self).setUp()

1548

1549

def test_normal(self):

1550

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

1551

json_data = json.loads(output)

1552

self.assertDictEqual(json_data, self.expected_json)

1553

1554

def test_one_client(self):

1555

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

1556

json_data = json.loads(output)

1557

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

1558

self.assertDictEqual(json_data, expected_json)

1559

1560

1561

class TestPrintTableCmd(TestCmd):

1562

def test_normal(self):

1563

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

1564

expected_output = "\n".join((

1565

"Name Enabled Timeout Last Successful Check",

1566

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

1567

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

1568

))

1569

self.assertEqual(output, expected_output)

1570

1571

def test_verbose(self):

1572

output = PrintTableCmd(verbose=True).output(

1573

self.clients.values())

1574

columns = (

1575

(

1576

"Name ",

1577

"foo ",

1578

"barbar ",

1579

),(

1580

"Enabled ",

1581

"Yes ",

1582

"Yes ",

1583

),(

1584

"Timeout ",

1585

"00:05:00 ",

1586

"00:05:00 ",

1587

),(

1588

"Last Successful Check ",

1589

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

1590

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

1591

),(

1592

"Created ",

1593

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

1594

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

1595

),(

1596

"Interval ",

1597

"00:02:00 ",

1598

"00:02:00 ",

1599

),(

1600

"Host ",

1601

"foo.example.org ",

1602

"192.0.2.3 ",

1603

),(

1604

("Key ID "

1605

" "),

1606

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1607

"ea90a27cddb12 "),

1608

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1609

"46c2b020f441c "),

1610

),(

1611

"Fingerprint ",

1612

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1613

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1614

),(

1615

"Check Is Running ",

1616

"No ",

1617

"Yes ",

1618

),(

1619

"Last Enabled ",

1620

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

1621

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

1622

),(

1623

"Approval Is Pending ",

1624

"No ",

1625

"No ",

1626

),(

1627

"Approved By Default ",

1628

"Yes ",

1629

"No ",

1630

),(

1631

"Last Approval Request ",

1632

" ",

1633

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

1634

),(

1635

"Approval Delay ",

1636

"00:00:00 ",

1637

"00:00:30 ",

1638

),(

1639

"Approval Duration ",

1640

"00:00:01 ",

1641

"1T02:03:05 ",

1642

),(

1643

"Checker ",

1644

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

1645

": ",

1646

),(

1647

"Extended Timeout ",

1648

"00:15:00 ",

1649

"00:15:00 ",

1650

),(

1651

"Expires ",

1652

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

1653

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

1654

),(

1655

"Last Checker Status",

1656

"0 ",

1657

"-2 ",

1658

)

1659

)

1660

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

1661

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

1662

for rows in columns)

1663

for line in range(num_lines))

1664

self.assertEqual(output, expected_output)

1665

1666

def test_one_client(self):

1667

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

1668

expected_output = "\n".join((

1669

"Name Enabled Timeout Last Successful Check",

1670

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

1671

))

1672

self.assertEqual(output, expected_output)

1673

1674

1675

class TestPropertyCmd(TestCmd):

1676

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

1677

def runTest(self):

1678

if not hasattr(self, "command"):

1679

return

1680

values_to_get = getattr(self, "values_to_get",

1681

self.values_to_set)

1682

for value_to_set, value_to_get in zip(self.values_to_set,

1683

values_to_get):

1684

for clientpath in self.clients:

1685

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

1686

old_value = client.attributes[self.propname]

1687

self.assertNotIsInstance(old_value, self.Unique)

1688

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

1689

self.run_command(value_to_set, self.clients)

1690

for clientpath in self.clients:

1691

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

1692

value = client.attributes[self.propname]

1693

self.assertNotIsInstance(value, self.Unique)

1694

self.assertEqual(value, value_to_get)

1695

1696

class Unique(object):

1697

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

1698

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

1699

1700

def run_command(self, value, clients):

1701

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

1702

1703

1704

class TestEnableCmd(TestPropertyCmd):

1705

command = EnableCmd

1706

propname = "Enabled"

1707

values_to_set = [dbus.Boolean(True)]

1708

1709

1710

class TestDisableCmd(TestPropertyCmd):

1711

command = DisableCmd

1712

propname = "Enabled"

1713

values_to_set = [dbus.Boolean(False)]

1714

1715

1716

class TestBumpTimeoutCmd(TestPropertyCmd):

1717

command = BumpTimeoutCmd

1718

propname = "LastCheckedOK"

1719

values_to_set = [""]

1720

1721

1722

class TestStartCheckerCmd(TestPropertyCmd):

1723

command = StartCheckerCmd

1724

propname = "CheckerRunning"

1725

values_to_set = [dbus.Boolean(True)]

1726

1727

1728

class TestStopCheckerCmd(TestPropertyCmd):

1729

command = StopCheckerCmd

1730

propname = "CheckerRunning"

1731

values_to_set = [dbus.Boolean(False)]

1732

1733

1734

class TestApproveByDefaultCmd(TestPropertyCmd):

1735

command = ApproveByDefaultCmd

1736

propname = "ApprovedByDefault"

1737

values_to_set = [dbus.Boolean(True)]

1738

1739

1740

class TestDenyByDefaultCmd(TestPropertyCmd):

1741

command = DenyByDefaultCmd

1742

propname = "ApprovedByDefault"

1743

values_to_set = [dbus.Boolean(False)]

1744

1745

1746

class TestPropertyValueCmd(TestPropertyCmd):

1747

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

1748

1749

def runTest(self):

1750

if type(self) is TestPropertyValueCmd:

1751

return

1752

return super(TestPropertyValueCmd, self).runTest()

1753

1754

def run_command(self, value, clients):

1755

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

1756

1757

1758

class TestSetCheckerCmd(TestPropertyValueCmd):

1759

command = SetCheckerCmd

1760

propname = "Checker"

1761

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

1762

1763

1764

class TestSetHostCmd(TestPropertyValueCmd):

1765

command = SetHostCmd

1766

propname = "Host"

1767

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

1768

1769

1770

class TestSetSecretCmd(TestPropertyValueCmd):

1771

command = SetSecretCmd

1772

propname = "Secret"

1773

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

1774

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

1775

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

1776

1777

1778

class TestSetTimeoutCmd(TestPropertyValueCmd):

1779

command = SetTimeoutCmd

1780

propname = "Timeout"

1781

values_to_set = [datetime.timedelta(),

1782

datetime.timedelta(minutes=5),

1783

datetime.timedelta(seconds=1),

1784

datetime.timedelta(weeks=1),

1785

datetime.timedelta(weeks=52)]

1786

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

1787

1788

1789

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1790

command = SetExtendedTimeoutCmd

1791

propname = "ExtendedTimeout"

1792

values_to_set = [datetime.timedelta(),

1793

datetime.timedelta(minutes=5),

1794

datetime.timedelta(seconds=1),

1795

datetime.timedelta(weeks=1),

1796

datetime.timedelta(weeks=52)]

1797

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

1798

1799

1800

class TestSetIntervalCmd(TestPropertyValueCmd):

1801

command = SetIntervalCmd

1802

propname = "Interval"

1803

values_to_set = [datetime.timedelta(),

1804

datetime.timedelta(minutes=5),

1805

datetime.timedelta(seconds=1),

1806

datetime.timedelta(weeks=1),

1807

datetime.timedelta(weeks=52)]

1808

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

1809

1810

1811

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1812

command = SetApprovalDelayCmd

1813

propname = "ApprovalDelay"

1814

values_to_set = [datetime.timedelta(),

1815

datetime.timedelta(minutes=5),

1816

datetime.timedelta(seconds=1),

1817

datetime.timedelta(weeks=1),

1818

datetime.timedelta(weeks=52)]

1819

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

1820

1821

1822

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1823

command = SetApprovalDurationCmd

1824

propname = "ApprovalDuration"

1825

values_to_set = [datetime.timedelta(),

1826

datetime.timedelta(minutes=5),

1827

datetime.timedelta(seconds=1),

1828

datetime.timedelta(weeks=1),

1829

datetime.timedelta(weeks=52)]

1830

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

1831

1832

1833

1834

def should_only_run_tests():

1835

parser = argparse.ArgumentParser(add_help=False)

1836

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

1837

args, unknown_args = parser.parse_known_args()

1838

run_tests = args.check

1839

if run_tests:

1840

# Remove --check argument from sys.argv

1841

sys.argv[1:] = unknown_args

1842

return run_tests

1843

1844

# Add all tests from doctest strings

1845

def load_tests(loader, tests, none):

1846

import doctest

1847

tests.addTests(doctest.DocTestSuite())

1848

return tests

341

1849

342

1850

if __name__ == "__main__":

343

main()

1851

try:

1852

if should_only_run_tests():

1853

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

1854

unittest.main()

1855

else:

1856

main()

1857

finally:

1858

logging.shutdown()