/mandos/trunk : revision 1070

1

#!/usr/bin/python

2

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

3

#

2

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

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

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

10

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

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

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

12

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

11

13

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

12

14

# (at your option) any later version.

13

15

#

14

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

15

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

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

16

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17

19

# GNU General Public License for more details.

18

#

20

#

19

21

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

20

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

21

#

22

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

23

#

22

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

23

#

24

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

25

#

24

26

25

27

from __future__ import (division, absolute_import, print_function,

26

28

unicode_literals)

27

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

28

35

import sys

29

import dbus

30

36

import argparse

31

37

import locale

32

38

import datetime

33

39

import re

34

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

35

64

36

65

locale.setlocale(locale.LC_ALL, "")

37

66

38

tablewords = {

39

"Name": "Name",

40

"Enabled": "Enabled",

41

"Timeout": "Timeout",

42

"LastCheckedOK": "Last Successful Check",

43

"LastApprovalRequest": "Last Approval Request",

44

"Created": "Created",

45

"Interval": "Interval",

46

"Host": "Host",

47

"Fingerprint": "Fingerprint",

48

"CheckerRunning": "Check Is Running",

49

"LastEnabled": "Last Enabled",

50

"ApprovalPending": "Approval Is Pending",

51

"ApprovedByDefault": "Approved By Default",

52

"ApprovalDelay": "Approval Delay",

53

"ApprovalDuration": "Approval Duration",

54

"Checker": "Checker",

55

}

56

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

57

domain = "se.bsnet.fukt"

58

busname = domain + ".Mandos"

59

server_path = "/"

60

server_interface = domain + ".Mandos"

61

client_interface = domain + ".Mandos.Client"

62

version = "1.3.1"

63

64

def timedelta_to_milliseconds(td):

65

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

66

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

67

+ (td.seconds * 1000)

68

+ (td.microseconds // 1000))

69

70

def milliseconds_to_string(ms):

71

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

72

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

73

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

74

"hours": td.seconds // 3600,

75

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

76

"seconds": td.seconds % 60,

77

})

78

79

def string_to_delta(interval):

80

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

81

82

>>> string_to_delta("7d")

83

datetime.timedelta(7)

84

>>> string_to_delta("60s")

85

datetime.timedelta(0, 60)

86

>>> string_to_delta("60m")

87

datetime.timedelta(0, 3600)

88

>>> string_to_delta("24h")

89

datetime.timedelta(1)

90

>>> string_to_delta("1w")

91

datetime.timedelta(7)

92

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

93

datetime.timedelta(0, 330)

94

"""

95

timevalue = datetime.timedelta(0)

96

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

97

98

for s in regexp.findall(interval):

99

try:

100

suffix = unicode(s[-1])

101

value = int(s[:-1])

102

if suffix == "d":

103

delta = datetime.timedelta(value)

104

elif suffix == "s":

105

delta = datetime.timedelta(0, value)

106

elif suffix == "m":

107

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

108

elif suffix == "h":

109

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

110

elif suffix == "w":

111

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

112

else:

113

raise ValueError

114

except (ValueError, IndexError):

115

raise ValueError

116

timevalue += delta

117

return timevalue

118

119

def print_clients(clients, keywords):

120

def valuetostring(value, keyword):

121

if type(value) is dbus.Boolean:

122

return "Yes" if value else "No"

123

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

124

"ApprovalDuration"):

125

return milliseconds_to_string(value)

126

return unicode(value)

127

128

# Create format string to print table rows

129

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

130

max(len(tablewords[key]),

131

max(len(valuetostring(client[key],

132

key))

133

for client in

134

clients))

135

for key in keywords)

136

# Print header line

137

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

138

for client in clients:

139

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

140

for key in keywords))

141

142

def has_actions(options):

143

return any((options.enable,

144

options.disable,

145

options.bump_timeout,

146

options.start_checker,

147

options.stop_checker,

148

options.is_enabled,

149

options.remove,

150

options.checker is not None,

151

options.timeout is not None,

152

options.interval is not None,

153

options.approved_by_default is not None,

154

options.approval_delay is not None,

155

options.approval_duration is not None,

156

options.host is not None,

157

options.secret is not None,

158

options.approve,

159

options.deny))

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

160

81

161

82

def main():

162

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

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

126

sys.exit(1)

127

128

# Run all commands on clients

129

commands = commands_from_options(options)

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

163

135

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

164

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

136

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

165

137

help="show version number and exit")

166

138

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

167

139

help="Select all clients")

168

140

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

169

141

help="Print all fields")

170

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

171

help="Enable client")

172

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

173

help="disable client")

142

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

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

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

146

help="Enable client")

147

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

148

action="store_true",

149

help="disable client")

174

150

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

175

151

help="Bump timeout for client")

176

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

177

help="Start checker for client")

178

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

179

help="Stop checker for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

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

154

action="store_true",

155

help="Start checker for client")

156

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

157

action="store_true",

158

help="Stop checker for client")

180

159

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

181

160

help="Check if client is enabled")

182

161

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

183

162

help="Remove client")

184

163

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

185

164

help="Set checker command for client")

186

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

165

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

187

166

help="Set timeout for client")

188

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

167

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

168

help="Set extended timeout for client")

169

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

189

170

help="Set checker interval for client")

190

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

191

default=None, dest="approved_by_default",

192

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

193

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

194

dest="approved_by_default",

195

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

196

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

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

175

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

176

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

179

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

180

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

197

181

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

198

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

182

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

199

183

help="Set duration of one client approval")

200

184

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

201

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

185

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

186

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

202

187

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

203

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

204

help="Approve any current client request")

205

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

206

help="Deny any current client request")

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

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

191

help="Approve any current client request")

192

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

193

help="Deny any current client request")

194

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

195

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

196

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

197

help="Run self-test")

207

198

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

208

options = parser.parse_args()

209

210

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

199

200

201

def string_to_delta(interval):

202

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

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

213

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

214

215

>>> rfc3339_duration_to_delta("P7D")

216

datetime.timedelta(7)

217

>>> rfc3339_duration_to_delta("PT60S")

218

datetime.timedelta(0, 60)

219

>>> rfc3339_duration_to_delta("PT60M")

220

datetime.timedelta(0, 3600)

221

>>> rfc3339_duration_to_delta("P60M")

222

datetime.timedelta(1680)

223

>>> rfc3339_duration_to_delta("PT24H")

224

datetime.timedelta(1)

225

>>> rfc3339_duration_to_delta("P1W")

226

datetime.timedelta(7)

227

>>> rfc3339_duration_to_delta("PT5M30S")

228

datetime.timedelta(0, 330)

229

>>> rfc3339_duration_to_delta("P1DT3M20S")

230

datetime.timedelta(1, 200)

231

>>> # Can not be empty:

232

>>> rfc3339_duration_to_delta("")

233

Traceback (most recent call last):

234

...

235

ValueError: Invalid RFC 3339 duration: u''

236

>>> # Must start with "P":

237

>>> rfc3339_duration_to_delta("1D")

238

Traceback (most recent call last):

239

...

240

ValueError: Invalid RFC 3339 duration: u'1D'

241

>>> # Must use correct order

242

>>> rfc3339_duration_to_delta("PT1S2M")

243

Traceback (most recent call last):

244

...

245

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

246

>>> # Time needs time marker

247

>>> rfc3339_duration_to_delta("P1H2S")

248

Traceback (most recent call last):

249

...

250

ValueError: Invalid RFC 3339 duration: u'P1H2S'

251

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

252

>>> rfc3339_duration_to_delta("P1D2W")

253

Traceback (most recent call last):

254

...

255

ValueError: Invalid RFC 3339 duration: u'P1D2W'

256

>>> rfc3339_duration_to_delta("P2W2H")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: u'P2W2H'

260

"""

261

262

# Parsing an RFC 3339 duration with regular expressions is not

263

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

264

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

265

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

266

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

267

# avoid excessive use of external libraries.

268

269

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

270

Token = collections.namedtuple("Token", (

271

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

272

# a "group" containing digits

273

"value", # datetime.timedelta or None

274

"followers")) # Tokens valid after this token

275

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

276

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

277

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

278

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

279

datetime.timedelta(seconds=1),

280

frozenset((token_end, )))

281

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

282

datetime.timedelta(minutes=1),

283

frozenset((token_second, token_end)))

284

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

285

datetime.timedelta(hours=1),

286

frozenset((token_minute, token_end)))

287

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

288

None,

289

frozenset((token_hour, token_minute,

290

token_second)))

291

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

292

datetime.timedelta(days=1),

293

frozenset((token_time, token_end)))

294

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

295

datetime.timedelta(weeks=4),

296

frozenset((token_day, token_end)))

297

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

298

datetime.timedelta(weeks=52),

299

frozenset((token_month, token_end)))

300

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

301

datetime.timedelta(weeks=1),

302

frozenset((token_end, )))

303

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

304

frozenset((token_year, token_month,

305

token_day, token_time,

306

token_week)))

307

# Define starting values:

308

# Value so far

309

value = datetime.timedelta()

310

found_token = None

311

# Following valid tokens

312

followers = frozenset((token_duration, ))

313

# String left to parse

314

s = duration

315

# Loop until end token is found

316

while found_token is not token_end:

317

# Search for any currently valid tokens

318

for token in followers:

319

match = token.regexp.match(s)

320

if match is not None:

321

# Token found

322

if token.value is not None:

323

# Value found, parse digits

324

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

325

# Add to value so far

326

value += factor * token.value

327

# Strip token from string

328

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

329

# Go to found token

330

found_token = token

331

# Set valid next tokens

332

followers = found_token.followers

333

break

334

else:

335

# No currently valid tokens were found

336

raise ValueError("Invalid RFC 3339 duration: {!r}"

337

.format(duration))

338

# End token found

339

return value

340

341

342

def parse_pre_1_6_1_interval(interval):

343

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

344

and return a datetime.timedelta

345

346

>>> parse_pre_1_6_1_interval('7d')

347

datetime.timedelta(7)

348

>>> parse_pre_1_6_1_interval('60s')

349

datetime.timedelta(0, 60)

350

>>> parse_pre_1_6_1_interval('60m')

351

datetime.timedelta(0, 3600)

352

>>> parse_pre_1_6_1_interval('24h')

353

datetime.timedelta(1)

354

>>> parse_pre_1_6_1_interval('1w')

355

datetime.timedelta(7)

356

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

357

datetime.timedelta(0, 330)

358

>>> parse_pre_1_6_1_interval('')

359

datetime.timedelta(0)

360

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

361

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

362

datetime.timedelta(2, 480, 18000)

363

364

"""

365

366

value = datetime.timedelta(0)

367

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

368

369

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

370

if suffix == "d":

371

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

372

elif suffix == "s":

373

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

374

elif suffix == "m":

375

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

376

elif suffix == "h":

377

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

378

elif suffix == "w":

379

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

380

elif suffix == "":

381

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

382

return value

383

384

385

def check_option_syntax(parser, options):

386

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

387

argparse"""

388

389

def has_actions(options):

390

return any((options.enable,

391

options.disable,

392

options.bump_timeout,

393

options.start_checker,

394

options.stop_checker,

395

options.is_enabled,

396

options.remove,

397

options.checker is not None,

398

options.timeout is not None,

399

options.extended_timeout is not None,

400

options.interval is not None,

401

options.approved_by_default is not None,

402

options.approval_delay is not None,

403

options.approval_duration is not None,

404

options.host is not None,

405

options.secret is not None,

406

options.approve,

407

options.deny))

408

409

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

211

410

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

212

411

if options.verbose and has_actions(options):

213

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

214

" --all.")

412

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

415

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

215

416

if options.all and not has_actions(options):

216

417

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

217

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

218

428

try:

219

bus = dbus.SystemBus()

220

mandos_dbus_objc = bus.get_object(busname, server_path)

429

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

430

dbus_busname, server_dbus_path)

431

mandos_dbus_object = bus.get_object(dbus_busname,

432

server_dbus_path)

221

433

except dbus.exceptions.DBusException:

222

print("Could not connect to Mandos server",

223

file=sys.stderr)

434

log.critical("Could not connect to Mandos server")

224

435

sys.exit(1)

225

226

mandos_serv = dbus.Interface(mandos_dbus_objc,

227

dbus_interface = server_interface)

228

229

#block stderr since dbus library prints to stderr

230

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

231

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

232

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

233

os.close(null)

436

437

return mandos_dbus_object

438

439

440

def get_managed_objects(object_manager):

441

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

442

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

234

443

try:

444

with SilenceLogger("dbus.proxies"):

445

managed_objects = object_manager.GetManagedObjects()

446

except dbus.exceptions.DBusException as e:

447

log.critical("Failed to access Mandos server through D-Bus:"

448

"\n%s", e)

449

sys.exit(1)

450

return managed_objects

451

452

453

class SilenceLogger(object):

454

"Simple context manager to silence a particular logger"

455

def __init__(self, loggername):

456

self.logger = logging.getLogger(loggername)

457

458

def __enter__(self):

459

self.logger.addFilter(self.nullfilter)

460

return self

461

462

class NullFilter(logging.Filter):

463

def filter(self, record):

464

return False

465

466

nullfilter = NullFilter()

467

468

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

469

self.logger.removeFilter(self.nullfilter)

470

471

472

def commands_from_options(options):

473

474

commands = []

475

476

if options.is_enabled:

477

commands.append(IsEnabledCmd())

478

479

if options.approve:

480

commands.append(ApproveCmd())

481

482

if options.deny:

483

commands.append(DenyCmd())

484

485

if options.remove:

486

commands.append(RemoveCmd())

487

488

if options.dump_json:

489

commands.append(DumpJSONCmd())

490

491

if options.enable:

492

commands.append(EnableCmd())

493

494

if options.disable:

495

commands.append(DisableCmd())

496

497

if options.bump_timeout:

498

commands.append(BumpTimeoutCmd())

499

500

if options.start_checker:

501

commands.append(StartCheckerCmd())

502

503

if options.stop_checker:

504

commands.append(StopCheckerCmd())

505

506

if options.approved_by_default is not None:

507

if options.approved_by_default:

508

commands.append(ApproveByDefaultCmd())

509

else:

510

commands.append(DenyByDefaultCmd())

511

512

if options.checker is not None:

513

commands.append(SetCheckerCmd(options.checker))

514

515

if options.host is not None:

516

commands.append(SetHostCmd(options.host))

517

518

if options.secret is not None:

519

commands.append(SetSecretCmd(options.secret))

520

521

if options.timeout is not None:

522

commands.append(SetTimeoutCmd(options.timeout))

523

524

if options.extended_timeout:

525

commands.append(

526

SetExtendedTimeoutCmd(options.extended_timeout))

527

528

if options.interval is not None:

529

commands.append(SetIntervalCmd(options.interval))

530

531

if options.approval_delay is not None:

532

commands.append(SetApprovalDelayCmd(options.approval_delay))

533

534

if options.approval_duration is not None:

535

commands.append(

536

SetApprovalDurationCmd(options.approval_duration))

537

538

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

539

# optionally verbosely

540

if not commands:

541

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

542

543

return commands

544

545

546

class Command(object):

547

"""Abstract class for commands"""

548

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

549

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

550

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

551

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

552

self.mandos = mandos

553

for clientpath, properties in clients.items():

554

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

555

dbus_busname, str(clientpath))

556

client = bus.get_object(dbus_busname, clientpath)

557

self.run_on_one_client(client, properties)

558

559

560

class IsEnabledCmd(Command):

561

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

562

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

563

if self.is_enabled(client, properties):

564

sys.exit(0)

565

sys.exit(1)

566

def is_enabled(self, client, properties):

567

return properties["Enabled"]

568

569

570

class ApproveCmd(Command):

571

def run_on_one_client(self, client, properties):

572

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

573

client.__dbus_object_path__, client_dbus_interface)

574

client.Approve(dbus.Boolean(True),

575

dbus_interface=client_dbus_interface)

576

577

578

class DenyCmd(Command):

579

def run_on_one_client(self, client, properties):

580

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

581

client.__dbus_object_path__, client_dbus_interface)

582

client.Approve(dbus.Boolean(False),

583

dbus_interface=client_dbus_interface)

584

585

586

class RemoveCmd(Command):

587

def run_on_one_client(self, client, properties):

588

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

589

server_dbus_path, server_dbus_interface,

590

str(client.__dbus_object_path__))

591

self.mandos.RemoveClient(client.__dbus_object_path__)

592

593

594

class OutputCmd(Command):

595

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

596

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

597

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

598

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

599

"ApprovalPending", "ApprovedByDefault",

600

"LastApprovalRequest", "ApprovalDelay",

601

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

602

"Expires", "LastCheckerStatus")

603

604

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

605

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

606

607

def output(self, clients):

608

raise NotImplementedError()

609

610

611

class DumpJSONCmd(OutputCmd):

612

def output(self, clients):

613

data = {client["Name"]:

614

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

615

for key in self.all_keywords}

616

for client in clients}

617

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

618

619

@staticmethod

620

def dbus_boolean_to_bool(value):

621

if isinstance(value, dbus.Boolean):

622

value = bool(value)

623

return value

624

625

626

class PrintTableCmd(OutputCmd):

627

def __init__(self, verbose=False):

628

self.verbose = verbose

629

630

def output(self, clients):

631

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

632

"LastCheckedOK")

633

keywords = default_keywords

634

if self.verbose:

635

keywords = self.all_keywords

636

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

637

638

class TableOfClients(object):

639

tableheaders = {

640

"Name": "Name",

641

"Enabled": "Enabled",

642

"Timeout": "Timeout",

643

"LastCheckedOK": "Last Successful Check",

644

"LastApprovalRequest": "Last Approval Request",

645

"Created": "Created",

646

"Interval": "Interval",

647

"Host": "Host",

648

"Fingerprint": "Fingerprint",

649

"KeyID": "Key ID",

650

"CheckerRunning": "Check Is Running",

651

"LastEnabled": "Last Enabled",

652

"ApprovalPending": "Approval Is Pending",

653

"ApprovedByDefault": "Approved By Default",

654

"ApprovalDelay": "Approval Delay",

655

"ApprovalDuration": "Approval Duration",

656

"Checker": "Checker",

657

"ExtendedTimeout": "Extended Timeout",

658

"Expires": "Expires",

659

"LastCheckerStatus": "Last Checker Status",

660

}

661

662

def __init__(self, clients, keywords):

663

self.clients = clients

664

self.keywords = keywords

665

666

def __str__(self):

667

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

668

669

if sys.version_info.major == 2:

670

__unicode__ = __str__

671

def __str__(self):

672

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

673

674

def rows(self):

675

format_string = self.row_formatting_string()

676

rows = [self.header_line(format_string)]

677

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

678

for client in self.clients)

679

return rows

680

681

def row_formatting_string(self):

682

"Format string used to format table rows"

683

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

684

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

685

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

686

for client in self.clients)),

687

key=key)

688

for key in self.keywords)

689

690

def string_from_client(self, client, key):

691

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

692

693

@classmethod

694

def valuetostring(cls, value, keyword):

695

if isinstance(value, dbus.Boolean):

696

return "Yes" if value else "No"

697

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

698

"ApprovalDuration", "ExtendedTimeout"):

699

return cls.milliseconds_to_string(value)

700

return str(value)

701

702

def header_line(self, format_string):

703

return format_string.format(**self.tableheaders)

704

705

def client_line(self, client, format_string):

706

return format_string.format(

707

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

708

for key in self.keywords})

709

710

@staticmethod

711

def milliseconds_to_string(ms):

712

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

713

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

714

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

715

if td.days else "",

716

hours=td.seconds // 3600,

717

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

718

seconds=td.seconds % 60))

719

720

721

class PropertyCmd(Command):

722

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

723

724

def run_on_one_client(self, client, properties):

725

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

726

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

727

client.__dbus_object_path__,

728

dbus.PROPERTIES_IFACE, client_dbus_interface,

729

self.propname, self.value_to_set

730

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

731

else bool(self.value_to_set))

732

client.Set(client_dbus_interface, self.propname,

733

self.value_to_set,

734

dbus_interface=dbus.PROPERTIES_IFACE)

735

736

@property

737

def propname(self):

738

raise NotImplementedError()

739

740

741

class EnableCmd(PropertyCmd):

742

propname = "Enabled"

743

value_to_set = dbus.Boolean(True)

744

745

746

class DisableCmd(PropertyCmd):

747

propname = "Enabled"

748

value_to_set = dbus.Boolean(False)

749

750

751

class BumpTimeoutCmd(PropertyCmd):

752

propname = "LastCheckedOK"

753

value_to_set = ""

754

755

756

class StartCheckerCmd(PropertyCmd):

757

propname = "CheckerRunning"

758

value_to_set = dbus.Boolean(True)

759

760

761

class StopCheckerCmd(PropertyCmd):

762

propname = "CheckerRunning"

763

value_to_set = dbus.Boolean(False)

764

765

766

class ApproveByDefaultCmd(PropertyCmd):

767

propname = "ApprovedByDefault"

768

value_to_set = dbus.Boolean(True)

769

770

771

class DenyByDefaultCmd(PropertyCmd):

772

propname = "ApprovedByDefault"

773

value_to_set = dbus.Boolean(False)

774

775

776

class PropertyValueCmd(PropertyCmd):

777

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

778

def __init__(self, value):

779

self.value_to_set = value

780

781

782

class SetCheckerCmd(PropertyValueCmd):

783

propname = "Checker"

784

785

786

class SetHostCmd(PropertyValueCmd):

787

propname = "Host"

788

789

790

class SetSecretCmd(PropertyValueCmd):

791

propname = "Secret"

792

793

@property

794

def value_to_set(self):

795

return self._vts

796

797

@value_to_set.setter

798

def value_to_set(self, value):

799

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

800

self._vts = value.read()

801

value.close()

802

803

804

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

805

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

806

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

807

808

@property

809

def value_to_set(self):

810

return self._vts

811

812

@value_to_set.setter

813

def value_to_set(self, value):

814

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

815

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

816

817

818

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

819

propname = "Timeout"

820

821

822

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

823

propname = "ExtendedTimeout"

824

825

826

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

827

propname = "Interval"

828

829

830

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

831

propname = "ApprovalDelay"

832

833

834

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

835

propname = "ApprovalDuration"

836

837

838

839

class Test_string_to_delta(unittest.TestCase):

840

def test_handles_basic_rfc3339(self):

841

self.assertEqual(string_to_delta("PT0S"),

842

datetime.timedelta())

843

self.assertEqual(string_to_delta("P0D"),

844

datetime.timedelta())

845

self.assertEqual(string_to_delta("PT1S"),

846

datetime.timedelta(0, 1))

847

self.assertEqual(string_to_delta("PT2H"),

848

datetime.timedelta(0, 7200))

849

850

def test_falls_back_to_pre_1_6_1_with_warning(self):

851

# assertLogs only exists in Python 3.4

852

if hasattr(self, "assertLogs"):

853

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

854

value = string_to_delta("2h")

855

else:

856

class WarningFilter(logging.Filter):

857

"""Don't show, but record the presence of, warnings"""

858

def filter(self, record):

859

is_warning = record.levelno >= logging.WARNING

860

self.found = is_warning or getattr(self, "found",

861

False)

862

return not is_warning

863

warning_filter = WarningFilter()

864

log.addFilter(warning_filter)

865

try:

866

value = string_to_delta("2h")

867

finally:

868

log.removeFilter(warning_filter)

869

self.assertTrue(getattr(warning_filter, "found", False))

870

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

871

872

873

class Test_check_option_syntax(unittest.TestCase):

874

def setUp(self):

875

self.parser = argparse.ArgumentParser()

876

add_command_line_options(self.parser)

877

878

def test_actions_requires_client_or_all(self):

879

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

880

options = self.parser.parse_args()

881

setattr(options, action, value)

882

with self.assertParseError():

883

self.check_option_syntax(options)

884

885

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

886

# check_option_syntax()

887

actions = {

888

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

889

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

890

# never used

891

"enable": True,

892

"disable": True,

893

"bump_timeout": True,

894

"start_checker": True,

895

"stop_checker": True,

896

"is_enabled": True,

897

"remove": True,

898

"checker": "x",

899

"timeout": datetime.timedelta(),

900

"extended_timeout": datetime.timedelta(),

901

"interval": datetime.timedelta(),

902

"approved_by_default": True,

903

"approval_delay": datetime.timedelta(),

904

"approval_duration": datetime.timedelta(),

905

"host": "x",

906

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

907

"approve": True,

908

"deny": True,

909

}

910

911

@contextlib.contextmanager

912

def assertParseError(self):

913

with self.assertRaises(SystemExit) as e:

914

with self.temporarily_suppress_stderr():

915

yield

916

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

917

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

918

# /argparse.html#exiting-methods

919

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

920

921

@staticmethod

922

@contextlib.contextmanager

923

def temporarily_suppress_stderr():

924

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

925

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

926

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

927

os.close(null)

235

928

try:

236

mandos_clients = mandos_serv.GetAllClientsWithProperties()

929

yield

237

930

finally:

238

#restore stderr

931

# restore stderr

239

932

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

240

933

os.close(stderrcopy)

241

except dbus.exceptions.DBusException, e:

242

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

243

file=sys.stderr)

244

sys.exit(1)

245

246

# Compile dict of (clients: properties) to process

247

clients={}

248

249

if options.all or not options.client:

250

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

251

for path, properties in

252

mandos_clients.iteritems())

253

else:

254

for name in options.client:

255

for path, client in mandos_clients.iteritems():

256

if client["Name"] == name:

257

client_objc = bus.get_object(busname, path)

258

clients[client_objc] = client

259

break

260

else:

261

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

262

file=sys.stderr)

263

sys.exit(1)

264

265

if not has_actions(options) and clients:

266

if options.verbose:

267

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

268

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

269

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

270

"LastEnabled", "ApprovalPending",

271

"ApprovedByDefault",

272

"LastApprovalRequest", "ApprovalDelay",

273

"ApprovalDuration", "Checker")

274

else:

275

keywords = defaultkeywords

276

277

print_clients(clients.values(), keywords)

278

else:

279

# Process each client in the list by all selected options

280

for client in clients:

281

if options.remove:

282

mandos_serv.RemoveClient(client.__dbus_object_path__)

283

if options.enable:

284

client.Enable(dbus_interface=client_interface)

285

if options.disable:

286

client.Disable(dbus_interface=client_interface)

287

if options.bump_timeout:

288

client.CheckedOK(dbus_interface=client_interface)

289

if options.start_checker:

290

client.StartChecker(dbus_interface=client_interface)

291

if options.stop_checker:

292

client.StopChecker(dbus_interface=client_interface)

293

if options.is_enabled:

294

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

295

"Enabled",

296

dbus_interface=

297

dbus.PROPERTIES_IFACE)

298

else 1)

299

if options.checker:

300

client.Set(client_interface, "Checker",

301

options.checker,

302

dbus_interface=dbus.PROPERTIES_IFACE)

303

if options.host:

304

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

305

dbus_interface=dbus.PROPERTIES_IFACE)

306

if options.interval:

307

client.Set(client_interface, "Interval",

308

timedelta_to_milliseconds

309

(string_to_delta(options.interval)),

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

if options.approval_delay:

312

client.Set(client_interface, "ApprovalDelay",

313

timedelta_to_milliseconds

314

(string_to_delta(options.

315

approval_delay)),

316

dbus_interface=dbus.PROPERTIES_IFACE)

317

if options.approval_duration:

318

client.Set(client_interface, "ApprovalDuration",

319

timedelta_to_milliseconds

320

(string_to_delta(options.

321

approval_duration)),

322

dbus_interface=dbus.PROPERTIES_IFACE)

323

if options.timeout:

324

client.Set(client_interface, "Timeout",

325

timedelta_to_milliseconds

326

(string_to_delta(options.timeout)),

327

dbus_interface=dbus.PROPERTIES_IFACE)

328

if options.secret:

329

client.Set(client_interface, "Secret",

330

dbus.ByteArray(open(options.secret,

331

"rb").read()),

332

dbus_interface=dbus.PROPERTIES_IFACE)

333

if options.approved_by_default is not None:

334

client.Set(client_interface, "ApprovedByDefault",

335

dbus.Boolean(options

336

.approved_by_default),

337

dbus_interface=dbus.PROPERTIES_IFACE)

338

if options.approve:

339

client.Approve(dbus.Boolean(True),

340

dbus_interface=client_interface)

341

elif options.deny:

342

client.Approve(dbus.Boolean(False),

343

dbus_interface=client_interface)

934

935

def check_option_syntax(self, options):

936

check_option_syntax(self.parser, options)

937

938

def test_actions_conflicts_with_verbose(self):

939

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

940

options = self.parser.parse_args()

941

setattr(options, action, value)

942

options.verbose = True

943

with self.assertParseError():

944

self.check_option_syntax(options)

945

946

def test_dump_json_conflicts_with_verbose(self):

947

options = self.parser.parse_args()

948

options.dump_json = True

949

options.verbose = True

950

with self.assertParseError():

951

self.check_option_syntax(options)

952

953

def test_dump_json_conflicts_with_action(self):

954

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

955

options = self.parser.parse_args()

956

setattr(options, action, value)

957

options.dump_json = True

958

with self.assertParseError():

959

self.check_option_syntax(options)

960

961

def test_all_can_not_be_alone(self):

962

options = self.parser.parse_args()

963

options.all = True

964

with self.assertParseError():

965

self.check_option_syntax(options)

966

967

def test_all_is_ok_with_any_action(self):

968

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

969

options = self.parser.parse_args()

970

setattr(options, action, value)

971

options.all = True

972

self.check_option_syntax(options)

973

974

def test_is_enabled_fails_without_client(self):

975

options = self.parser.parse_args()

976

options.is_enabled = True

977

with self.assertParseError():

978

self.check_option_syntax(options)

979

980

def test_is_enabled_works_with_one_client(self):

981

options = self.parser.parse_args()

982

options.is_enabled = True

983

options.client = ["foo"]

984

self.check_option_syntax(options)

985

986

def test_is_enabled_fails_with_two_clients(self):

987

options = self.parser.parse_args()

988

options.is_enabled = True

989

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

990

with self.assertParseError():

991

self.check_option_syntax(options)

992

993

def test_remove_can_only_be_combined_with_action_deny(self):

994

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

995

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

996

continue

997

options = self.parser.parse_args()

998

setattr(options, action, value)

999

options.all = True

1000

options.remove = True

1001

with self.assertParseError():

1002

self.check_option_syntax(options)

1003

1004

1005

class Test_get_mandos_dbus_object(unittest.TestCase):

1006

def test_calls_and_returns_get_object_on_bus(self):

1007

class MockBus(object):

1008

called = False

1009

def get_object(mockbus_self, busname, dbus_path):

1010

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

1011

# this MockBus instance is in "mockbus_self".

1012

self.assertEqual(busname, dbus_busname)

1013

self.assertEqual(dbus_path, server_dbus_path)

1014

mockbus_self.called = True

1015

return mockbus_self

1016

1017

mockbus = get_mandos_dbus_object(bus=MockBus())

1018

self.assertIsInstance(mockbus, MockBus)

1019

self.assertTrue(mockbus.called)

1020

1021

def test_logs_and_exits_on_dbus_error(self):

1022

class MockBusFailing(object):

1023

def get_object(self, busname, dbus_path):

1024

raise dbus.exceptions.DBusException("Test")

1025

1026

# assertLogs only exists in Python 3.4

1027

if hasattr(self, "assertLogs"):

1028

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

1029

with self.assertRaises(SystemExit) as e:

1030

bus = get_mandos_dbus_object(bus=MockBus())

1031

else:

1032

critical_filter = self.CriticalFilter()

1033

log.addFilter(critical_filter)

1034

try:

1035

with self.assertRaises(SystemExit) as e:

1036

get_mandos_dbus_object(bus=MockBusFailing())

1037

finally:

1038

log.removeFilter(critical_filter)

1039

self.assertTrue(critical_filter.found)

1040

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

1041

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

1042

else:

1043

self.assertIsNotNone(e.exception.code)

1044

1045

class CriticalFilter(logging.Filter):

1046

"""Don't show, but register, critical messages"""

1047

found = False

1048

def filter(self, record):

1049

is_critical = record.levelno >= logging.CRITICAL

1050

self.found = is_critical or self.found

1051

return not is_critical

1052

1053

1054

class Test_get_managed_objects(unittest.TestCase):

1055

def test_calls_and_returns_GetManagedObjects(self):

1056

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

1057

class MockObjectManager(object):

1058

@staticmethod

1059

def GetManagedObjects():

1060

return managed_objects

1061

retval = get_managed_objects(MockObjectManager())

1062

self.assertDictEqual(managed_objects, retval)

1063

1064

def test_logs_and_exits_on_dbus_error(self):

1065

class MockObjectManagerFailing(object):

1066

@staticmethod

1067

def GetManagedObjects():

1068

raise dbus.exceptions.DBusException("Test")

1069

1070

if hasattr(self, "assertLogs"):

1071

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

1072

with self.assertRaises(SystemExit):

1073

get_managed_objects(MockObjectManagerFailing())

1074

else:

1075

critical_filter = self.CriticalFilter()

1076

log.addFilter(critical_filter)

1077

try:

1078

with self.assertRaises(SystemExit) as e:

1079

get_managed_objects(MockObjectManagerFailing())

1080

finally:

1081

log.removeFilter(critical_filter)

1082

self.assertTrue(critical_filter.found)

1083

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

1084

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

1085

else:

1086

self.assertIsNotNone(e.exception.code)

1087

1088

class CriticalFilter(logging.Filter):

1089

"""Don't show, but register, critical messages"""

1090

found = False

1091

def filter(self, record):

1092

is_critical = record.levelno >= logging.CRITICAL

1093

self.found = is_critical or self.found

1094

return not is_critical

1095

1096

1097

class Test_SilenceLogger(unittest.TestCase):

1098

loggername = "mandos-ctl.Test_SilenceLogger"

1099

log = logging.getLogger(loggername)

1100

log.propagate = False

1101

log.addHandler(logging.NullHandler())

1102

1103

def setUp(self):

1104

self.counting_filter = self.CountingFilter()

1105

1106

class CountingFilter(logging.Filter):

1107

"Count number of records"

1108

count = 0

1109

def filter(self, record):

1110

self.count += 1

1111

return True

1112

1113

def test_should_filter_records_only_when_active(self):

1114

try:

1115

with SilenceLogger(self.loggername):

1116

self.log.addFilter(self.counting_filter)

1117

self.log.info("Filtered log message 1")

1118

self.log.info("Non-filtered message 2")

1119

self.log.info("Non-filtered message 3")

1120

finally:

1121

self.log.removeFilter(self.counting_filter)

1122

self.assertEqual(self.counting_filter.count, 2)

1123

1124

1125

class Test_commands_from_options(unittest.TestCase):

1126

def setUp(self):

1127

self.parser = argparse.ArgumentParser()

1128

add_command_line_options(self.parser)

1129

1130

def test_is_enabled(self):

1131

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

1132

IsEnabledCmd)

1133

1134

def assert_command_from_args(self, args, command_cls,

1135

**cmd_attrs):

1136

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

1137

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

1138

options = self.parser.parse_args(args)

1139

check_option_syntax(self.parser, options)

1140

commands = commands_from_options(options)

1141

self.assertEqual(len(commands), 1)

1142

command = commands[0]

1143

self.assertIsInstance(command, command_cls)

1144

for key, value in cmd_attrs.items():

1145

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

1146

1147

def test_is_enabled_short(self):

1148

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

1149

1150

def test_approve(self):

1151

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

1152

ApproveCmd)

1153

1154

def test_approve_short(self):

1155

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

1156

1157

def test_deny(self):

1158

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

1159

1160

def test_deny_short(self):

1161

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

1162

1163

def test_remove(self):

1164

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

1165

RemoveCmd)

1166

1167

def test_deny_before_remove(self):

1168

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

1169

"foo"])

1170

check_option_syntax(self.parser, options)

1171

commands = commands_from_options(options)

1172

self.assertEqual(len(commands), 2)

1173

self.assertIsInstance(commands[0], DenyCmd)

1174

self.assertIsInstance(commands[1], RemoveCmd)

1175

1176

def test_deny_before_remove_reversed(self):

1177

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

1178

"--all"])

1179

check_option_syntax(self.parser, options)

1180

commands = commands_from_options(options)

1181

self.assertEqual(len(commands), 2)

1182

self.assertIsInstance(commands[0], DenyCmd)

1183

self.assertIsInstance(commands[1], RemoveCmd)

1184

1185

def test_remove_short(self):

1186

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

1187

1188

def test_dump_json(self):

1189

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

1190

1191

def test_enable(self):

1192

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

1193

1194

def test_enable_short(self):

1195

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

1196

1197

def test_disable(self):

1198

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

1199

DisableCmd)

1200

1201

def test_disable_short(self):

1202

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

1203

1204

def test_bump_timeout(self):

1205

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

1206

BumpTimeoutCmd)

1207

1208

def test_bump_timeout_short(self):

1209

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

1210

1211

def test_start_checker(self):

1212

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

1213

StartCheckerCmd)

1214

1215

def test_stop_checker(self):

1216

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

1217

StopCheckerCmd)

1218

1219

def test_approve_by_default(self):

1220

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

1221

ApproveByDefaultCmd)

1222

1223

def test_deny_by_default(self):

1224

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

1225

DenyByDefaultCmd)

1226

1227

def test_checker(self):

1228

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

1229

SetCheckerCmd, value_to_set=":")

1230

1231

def test_checker_empty(self):

1232

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

1233

SetCheckerCmd, value_to_set="")

1234

1235

def test_checker_short(self):

1236

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

1237

SetCheckerCmd, value_to_set=":")

1238

1239

def test_host(self):

1240

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

1241

"foo"], SetHostCmd,

1242

value_to_set="foo.example.org")

1243

1244

def test_host_short(self):

1245

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

1246

"foo"], SetHostCmd,

1247

value_to_set="foo.example.org")

1248

1249

def test_secret_devnull(self):

1250

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

1251

"foo"], SetSecretCmd,

1252

value_to_set=b"")

1253

1254

def test_secret_tempfile(self):

1255

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

1256

value = b"secret\0xyzzy\nbar"

1257

f.write(value)

1258

f.seek(0)

1259

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

1260

"foo"], SetSecretCmd,

1261

value_to_set=value)

1262

1263

def test_secret_devnull_short(self):

1264

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

1265

SetSecretCmd, value_to_set=b"")

1266

1267

def test_secret_tempfile_short(self):

1268

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

1269

value = b"secret\0xyzzy\nbar"

1270

f.write(value)

1271

f.seek(0)

1272

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

1273

SetSecretCmd,

1274

value_to_set=value)

1275

1276

def test_timeout(self):

1277

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

1278

SetTimeoutCmd,

1279

value_to_set=300000)

1280

1281

def test_timeout_short(self):

1282

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

1283

SetTimeoutCmd,

1284

value_to_set=300000)

1285

1286

def test_extended_timeout(self):

1287

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

1288

"foo"],

1289

SetExtendedTimeoutCmd,

1290

value_to_set=900000)

1291

1292

def test_interval(self):

1293

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

1294

SetIntervalCmd,

1295

value_to_set=120000)

1296

1297

def test_interval_short(self):

1298

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

1299

SetIntervalCmd,

1300

value_to_set=120000)

1301

1302

def test_approval_delay(self):

1303

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

1304

"foo"], SetApprovalDelayCmd,

1305

value_to_set=30000)

1306

1307

def test_approval_duration(self):

1308

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

1309

"foo"], SetApprovalDurationCmd,

1310

value_to_set=1000)

1311

1312

def test_print_table(self):

1313

self.assert_command_from_args([], PrintTableCmd,

1314

verbose=False)

1315

1316

def test_print_table_verbose(self):

1317

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

1318

verbose=True)

1319

1320

def test_print_table_verbose_short(self):

1321

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

1322

verbose=True)

1323

1324

1325

class TestCmd(unittest.TestCase):

1326

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

1327

1328

def setUp(self):

1329

testcase = self

1330

class MockClient(object):

1331

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

1332

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

1333

self.attributes = attributes

1334

self.attributes["Name"] = name

1335

self.calls = []

1336

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

1337

testcase.assertEqual(interface, client_dbus_interface)

1338

testcase.assertEqual(dbus_interface,

1339

dbus.PROPERTIES_IFACE)

1340

self.attributes[propname] = value

1341

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

1342

testcase.assertEqual(interface, client_dbus_interface)

1343

testcase.assertEqual(dbus_interface,

1344

dbus.PROPERTIES_IFACE)

1345

return self.attributes[propname]

1346

def Approve(self, approve, dbus_interface):

1347

testcase.assertEqual(dbus_interface,

1348

client_dbus_interface)

1349

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

1350

dbus_interface)))

1351

self.client = MockClient(

1352

"foo",

1353

KeyID=("92ed150794387c03ce684574b1139a65"

1354

"94a34f895daaaf09fd8ea90a27cddb12"),

1355

Secret=b"secret",

1356

Host="foo.example.org",

1357

Enabled=dbus.Boolean(True),

1358

Timeout=300000,

1359

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

1360

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

1361

Interval=120000,

1362

Fingerprint=("778827225BA7DE539C5A"

1363

"7CFA59CFF7CDBD9A5920"),

1364

CheckerRunning=dbus.Boolean(False),

1365

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

1366

ApprovalPending=dbus.Boolean(False),

1367

ApprovedByDefault=dbus.Boolean(True),

1368

LastApprovalRequest="",

1369

ApprovalDelay=0,

1370

ApprovalDuration=1000,

1371

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

1372

ExtendedTimeout=900000,

1373

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

1374

LastCheckerStatus=0)

1375

self.other_client = MockClient(

1376

"barbar",

1377

KeyID=("0558568eedd67d622f5c83b35a115f79"

1378

"6ab612cff5ad227247e46c2b020f441c"),

1379

Secret=b"secretbar",

1380

Host="192.0.2.3",

1381

Enabled=dbus.Boolean(True),

1382

Timeout=300000,

1383

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

1384

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

1385

Interval=120000,

1386

Fingerprint=("3E393AEAEFB84C7E89E2"

1387

"F547B3A107558FCA3A27"),

1388

CheckerRunning=dbus.Boolean(True),

1389

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

1390

ApprovalPending=dbus.Boolean(False),

1391

ApprovedByDefault=dbus.Boolean(False),

1392

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

1393

ApprovalDelay=30000,

1394

ApprovalDuration=93785000,

1395

Checker=":",

1396

ExtendedTimeout=900000,

1397

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

1398

LastCheckerStatus=-2)

1399

self.clients = collections.OrderedDict(

1400

[

1401

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

1402

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

1403

])

1404

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

1405

1406

@property

1407

def bus(self):

1408

class Bus(object):

1409

@staticmethod

1410

def get_object(client_bus_name, path):

1411

self.assertEqual(client_bus_name, dbus_busname)

1412

return {

1413

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

1414

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

1415

"/clients/foo": self.client,

1416

"/clients/barbar": self.other_client,

1417

}[path]

1418

return Bus()

1419

1420

1421

class TestIsEnabledCmd(TestCmd):

1422

def test_is_enabled(self):

1423

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

1424

properties)

1425

for client, properties

1426

in self.clients.items()))

1427

1428

def test_is_enabled_run_exits_successfully(self):

1429

with self.assertRaises(SystemExit) as e:

1430

IsEnabledCmd().run(self.one_client)

1431

if e.exception.code is not None:

1432

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

1433

else:

1434

self.assertIsNone(e.exception.code)

1435

1436

def test_is_enabled_run_exits_with_failure(self):

1437

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

1438

with self.assertRaises(SystemExit) as e:

1439

IsEnabledCmd().run(self.one_client)

1440

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

1441

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

1442

else:

1443

self.assertIsNotNone(e.exception.code)

1444

1445

1446

class TestApproveCmd(TestCmd):

1447

def test_approve(self):

1448

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

1449

for clientpath in self.clients:

1450

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

1451

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

1452

client.calls)

1453

1454

1455

class TestDenyCmd(TestCmd):

1456

def test_deny(self):

1457

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

1458

for clientpath in self.clients:

1459

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

1460

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

1461

client.calls)

1462

1463

1464

class TestRemoveCmd(TestCmd):

1465

def test_remove(self):

1466

class MockMandos(object):

1467

def __init__(self):

1468

self.calls = []

1469

def RemoveClient(self, dbus_path):

1470

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

1471

mandos = MockMandos()

1472

super(TestRemoveCmd, self).setUp()

1473

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

1474

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

1475

for clientpath in self.clients:

1476

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

1477

mandos.calls)

1478

1479

1480

class TestDumpJSONCmd(TestCmd):

1481

def setUp(self):

1482

self.expected_json = {

1483

"foo": {

1484

"Name": "foo",

1485

"KeyID": ("92ed150794387c03ce684574b1139a65"

1486

"94a34f895daaaf09fd8ea90a27cddb12"),

1487

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

1488

"Enabled": True,

1489

"Timeout": 300000,

1490

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

1491

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

1492

"Interval": 120000,

1493

"Fingerprint": ("778827225BA7DE539C5A"

1494

"7CFA59CFF7CDBD9A5920"),

1495

"CheckerRunning": False,

1496

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

1497

"ApprovalPending": False,

1498

"ApprovedByDefault": True,

1499

"LastApprovalRequest": "",

1500

"ApprovalDelay": 0,

1501

"ApprovalDuration": 1000,

1502

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

1503

"ExtendedTimeout": 900000,

1504

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

1505

"LastCheckerStatus": 0,

1506

},

1507

"barbar": {

1508

"Name": "barbar",

1509

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1510

"6ab612cff5ad227247e46c2b020f441c"),

1511

"Host": "192.0.2.3",

1512

"Enabled": True,

1513

"Timeout": 300000,

1514

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

1515

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

1516

"Interval": 120000,

1517

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1518

"F547B3A107558FCA3A27"),

1519

"CheckerRunning": True,

1520

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

1521

"ApprovalPending": False,

1522

"ApprovedByDefault": False,

1523

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

1524

"ApprovalDelay": 30000,

1525

"ApprovalDuration": 93785000,

1526

"Checker": ":",

1527

"ExtendedTimeout": 900000,

1528

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

1529

"LastCheckerStatus": -2,

1530

},

1531

}

1532

return super(TestDumpJSONCmd, self).setUp()

1533

1534

def test_normal(self):

1535

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

1536

json_data = json.loads(output)

1537

self.assertDictEqual(json_data, self.expected_json)

1538

1539

def test_one_client(self):

1540

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

1541

json_data = json.loads(output)

1542

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

1543

self.assertDictEqual(json_data, expected_json)

1544

1545

1546

class TestPrintTableCmd(TestCmd):

1547

def test_normal(self):

1548

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

1549

expected_output = "\n".join((

1550

"Name Enabled Timeout Last Successful Check",

1551

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

1552

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

1553

))

1554

self.assertEqual(output, expected_output)

1555

1556

def test_verbose(self):

1557

output = PrintTableCmd(verbose=True).output(

1558

self.clients.values())

1559

columns = (

1560

(

1561

"Name ",

1562

"foo ",

1563

"barbar ",

1564

),(

1565

"Enabled ",

1566

"Yes ",

1567

"Yes ",

1568

),(

1569

"Timeout ",

1570

"00:05:00 ",

1571

"00:05:00 ",

1572

),(

1573

"Last Successful Check ",

1574

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

1575

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

1576

),(

1577

"Created ",

1578

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

1579

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

1580

),(

1581

"Interval ",

1582

"00:02:00 ",

1583

"00:02:00 ",

1584

),(

1585

"Host ",

1586

"foo.example.org ",

1587

"192.0.2.3 ",

1588

),(

1589

("Key ID "

1590

" "),

1591

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1592

"ea90a27cddb12 "),

1593

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1594

"46c2b020f441c "),

1595

),(

1596

"Fingerprint ",

1597

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1598

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1599

),(

1600

"Check Is Running ",

1601

"No ",

1602

"Yes ",

1603

),(

1604

"Last Enabled ",

1605

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

1606

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

1607

),(

1608

"Approval Is Pending ",

1609

"No ",

1610

"No ",

1611

),(

1612

"Approved By Default ",

1613

"Yes ",

1614

"No ",

1615

),(

1616

"Last Approval Request ",

1617

" ",

1618

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

1619

),(

1620

"Approval Delay ",

1621

"00:00:00 ",

1622

"00:00:30 ",

1623

),(

1624

"Approval Duration ",

1625

"00:00:01 ",

1626

"1T02:03:05 ",

1627

),(

1628

"Checker ",

1629

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

1630

": ",

1631

),(

1632

"Extended Timeout ",

1633

"00:15:00 ",

1634

"00:15:00 ",

1635

),(

1636

"Expires ",

1637

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

1638

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

1639

),(

1640

"Last Checker Status",

1641

"0 ",

1642

"-2 ",

1643

)

1644

)

1645

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

1646

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

1647

for rows in columns)

1648

for line in range(num_lines))

1649

self.assertEqual(output, expected_output)

1650

1651

def test_one_client(self):

1652

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

1653

expected_output = "\n".join((

1654

"Name Enabled Timeout Last Successful Check",

1655

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

1656

))

1657

self.assertEqual(output, expected_output)

1658

1659

1660

class TestPropertyCmd(TestCmd):

1661

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

1662

def runTest(self):

1663

if not hasattr(self, "command"):

1664

return

1665

values_to_get = getattr(self, "values_to_get",

1666

self.values_to_set)

1667

for value_to_set, value_to_get in zip(self.values_to_set,

1668

values_to_get):

1669

for clientpath in self.clients:

1670

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

1671

old_value = client.attributes[self.propname]

1672

self.assertNotIsInstance(old_value, self.Unique)

1673

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

1674

self.run_command(value_to_set, self.clients)

1675

for clientpath in self.clients:

1676

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

1677

value = client.attributes[self.propname]

1678

self.assertNotIsInstance(value, self.Unique)

1679

self.assertEqual(value, value_to_get)

1680

1681

class Unique(object):

1682

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

1683

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

1684

1685

def run_command(self, value, clients):

1686

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

1687

1688

1689

class TestEnableCmd(TestPropertyCmd):

1690

command = EnableCmd

1691

propname = "Enabled"

1692

values_to_set = [dbus.Boolean(True)]

1693

1694

1695

class TestDisableCmd(TestPropertyCmd):

1696

command = DisableCmd

1697

propname = "Enabled"

1698

values_to_set = [dbus.Boolean(False)]

1699

1700

1701

class TestBumpTimeoutCmd(TestPropertyCmd):

1702

command = BumpTimeoutCmd

1703

propname = "LastCheckedOK"

1704

values_to_set = [""]

1705

1706

1707

class TestStartCheckerCmd(TestPropertyCmd):

1708

command = StartCheckerCmd

1709

propname = "CheckerRunning"

1710

values_to_set = [dbus.Boolean(True)]

1711

1712

1713

class TestStopCheckerCmd(TestPropertyCmd):

1714

command = StopCheckerCmd

1715

propname = "CheckerRunning"

1716

values_to_set = [dbus.Boolean(False)]

1717

1718

1719

class TestApproveByDefaultCmd(TestPropertyCmd):

1720

command = ApproveByDefaultCmd

1721

propname = "ApprovedByDefault"

1722

values_to_set = [dbus.Boolean(True)]

1723

1724

1725

class TestDenyByDefaultCmd(TestPropertyCmd):

1726

command = DenyByDefaultCmd

1727

propname = "ApprovedByDefault"

1728

values_to_set = [dbus.Boolean(False)]

1729

1730

1731

class TestPropertyValueCmd(TestPropertyCmd):

1732

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

1733

1734

def runTest(self):

1735

if type(self) is TestPropertyValueCmd:

1736

return

1737

return super(TestPropertyValueCmd, self).runTest()

1738

1739

def run_command(self, value, clients):

1740

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

1741

1742

1743

class TestSetCheckerCmd(TestPropertyValueCmd):

1744

command = SetCheckerCmd

1745

propname = "Checker"

1746

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

1747

1748

1749

class TestSetHostCmd(TestPropertyValueCmd):

1750

command = SetHostCmd

1751

propname = "Host"

1752

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

1753

1754

1755

class TestSetSecretCmd(TestPropertyValueCmd):

1756

command = SetSecretCmd

1757

propname = "Secret"

1758

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

1759

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

1760

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

1761

1762

1763

class TestSetTimeoutCmd(TestPropertyValueCmd):

1764

command = SetTimeoutCmd

1765

propname = "Timeout"

1766

values_to_set = [datetime.timedelta(),

1767

datetime.timedelta(minutes=5),

1768

datetime.timedelta(seconds=1),

1769

datetime.timedelta(weeks=1),

1770

datetime.timedelta(weeks=52)]

1771

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

1772

1773

1774

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1775

command = SetExtendedTimeoutCmd

1776

propname = "ExtendedTimeout"

1777

values_to_set = [datetime.timedelta(),

1778

datetime.timedelta(minutes=5),

1779

datetime.timedelta(seconds=1),

1780

datetime.timedelta(weeks=1),

1781

datetime.timedelta(weeks=52)]

1782

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

1783

1784

1785

class TestSetIntervalCmd(TestPropertyValueCmd):

1786

command = SetIntervalCmd

1787

propname = "Interval"

1788

values_to_set = [datetime.timedelta(),

1789

datetime.timedelta(minutes=5),

1790

datetime.timedelta(seconds=1),

1791

datetime.timedelta(weeks=1),

1792

datetime.timedelta(weeks=52)]

1793

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

1794

1795

1796

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1797

command = SetApprovalDelayCmd

1798

propname = "ApprovalDelay"

1799

values_to_set = [datetime.timedelta(),

1800

datetime.timedelta(minutes=5),

1801

datetime.timedelta(seconds=1),

1802

datetime.timedelta(weeks=1),

1803

datetime.timedelta(weeks=52)]

1804

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

1805

1806

1807

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1808

command = SetApprovalDurationCmd

1809

propname = "ApprovalDuration"

1810

values_to_set = [datetime.timedelta(),

1811

datetime.timedelta(minutes=5),

1812

datetime.timedelta(seconds=1),

1813

datetime.timedelta(weeks=1),

1814

datetime.timedelta(weeks=52)]

1815

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

1816

1817

1818

1819

def should_only_run_tests():

1820

parser = argparse.ArgumentParser(add_help=False)

1821

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

1822

args, unknown_args = parser.parse_known_args()

1823

run_tests = args.check

1824

if run_tests:

1825

# Remove --check argument from sys.argv

1826

sys.argv[1:] = unknown_args

1827

return run_tests

1828

1829

# Add all tests from doctest strings

1830

def load_tests(loader, tests, none):

1831

import doctest

1832

tests.addTests(doctest.DocTestSuite())

1833

return tests

344

1834

345

1835

if __name__ == "__main__":

346

main()

1836

try:

1837

if should_only_run_tests():

1838

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

1839

unittest.main()

1840

else:

1841

main()

1842

finally:

1843

logging.shutdown()