/mandos/trunk : revision 1162

1

#!/usr/bin/python

2

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

3

4

from __future__ import division

2

# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

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

13

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

14

# (at your option) any later version.

15

#

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19

# GNU General Public License for more details.

20

#

21

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

22

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

23

#

24

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

25

#

26

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

5

35

import sys

6

import dbus

7

from optparse import OptionParser

36

import argparse

8

37

import locale

9

38

import datetime

10

39

import re

11

12

locale.setlocale(locale.LC_ALL, u'')

13

14

tablewords = {

15

'name': u'Name',

16

'enabled': u'Enabled',

17

'timeout': u'Timeout',

18

'last_checked_ok': u'Last Successful Check',

19

'created': u'Created',

20

'interval': u'Interval',

21

'host': u'Host',

22

'fingerprint': u'Fingerprint',

23

'checker_running': u'Check Is Running',

24

'last_enabled': u'Last Enabled',

25

'checker': u'Checker',

26

}

27

defaultkeywords = ('name', 'enabled', 'timeout', 'last_checked_ok')

28

domain = 'se.bsnet.fukt'

29

busname = domain + '.Mandos'

30

server_path = '/'

31

server_interface = domain + '.Mandos'

32

client_interface = domain + '.Mandos.Client'

33

version = "1.0.14"

34

bus = dbus.SystemBus()

35

mandos_dbus_objc = bus.get_object(busname, server_path)

36

mandos_serv = dbus.Interface(mandos_dbus_objc,

37

dbus_interface = server_interface)

38

mandos_clients = mandos_serv.GetAllClientsWithProperties()

39

40

def timedelta_to_milliseconds(td):

41

"Convert a datetime.timedelta object to milliseconds"

42

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

43

+ (td.seconds * 1000)

44

+ (td.microseconds // 1000))

45

46

def milliseconds_to_string(ms):

47

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

48

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

49

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

50

"hours": td.seconds // 3600,

51

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

52

"seconds": td.seconds % 60,

53

})

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

if sys.version_info.major == 2:

50

__metaclass__ = type

51

52

try:

53

import pydbus

54

import gi

55

dbus_python = None

56

except ImportError:

57

import dbus as dbus_python

58

pydbus = None

59

class gi:

60

"""Dummy gi module, for the tests"""

61

class repository:

62

class GLib:

63

class Error(Exception):

64

pass

65

66

# Show warnings by default

67

if not sys.warnoptions:

68

import warnings

69

warnings.simplefilter("default")

70

71

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

72

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

73

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

74

75

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

76

77

if sys.version_info.major == 2:

78

str = unicode

79

import StringIO

80

io.StringIO = StringIO.StringIO

81

82

locale.setlocale(locale.LC_ALL, "")

83

84

version = "1.8.8"

85

86

87

def main():

88

parser = argparse.ArgumentParser()

89

add_command_line_options(parser)

90

91

options = parser.parse_args()

92

check_option_syntax(parser, options)

93

94

clientnames = options.client

95

96

if options.debug:

97

log.setLevel(logging.DEBUG)

98

99

if pydbus is not None:

100

bus = pydbus_adapter.CachingBus(pydbus)

101

else:

102

bus = dbus_python_adapter.CachingBus(dbus_python)

103

104

try:

105

all_clients = bus.get_clients_and_properties()

106

except dbus.ConnectFailed as e:

107

log.critical("Could not connect to Mandos server: %s", e)

108

sys.exit(1)

109

except dbus.Error as e:

110

log.critical(

111

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

112

sys.exit(1)

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)

132

133

134

def add_command_line_options(parser):

135

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

136

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

137

help="show version number and exit")

138

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

139

help="Select all clients")

140

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

141

help="Print all fields")

142

parser.add_argument("-j", "--dump-json", dest="commands",

143

action="append_const", default=[],

144

const=command.DumpJSON(),

145

help="Dump client data in JSON format")

146

enable_disable = parser.add_mutually_exclusive_group()

147

enable_disable.add_argument("-e", "--enable", dest="commands",

148

action="append_const", default=[],

149

const=command.Enable(),

150

help="Enable client")

151

enable_disable.add_argument("-d", "--disable", dest="commands",

152

action="append_const", default=[],

153

const=command.Disable(),

154

help="disable client")

155

parser.add_argument("-b", "--bump-timeout", dest="commands",

156

action="append_const", default=[],

157

const=command.BumpTimeout(),

158

help="Bump timeout for client")

159

start_stop_checker = parser.add_mutually_exclusive_group()

160

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

161

dest="commands",

162

action="append_const", default=[],

163

const=command.StartChecker(),

164

help="Start checker for client")

165

start_stop_checker.add_argument("--stop-checker", dest="commands",

166

action="append_const", default=[],

167

const=command.StopChecker(),

168

help="Stop checker for client")

169

parser.add_argument("-V", "--is-enabled", dest="commands",

170

action="append_const", default=[],

171

const=command.IsEnabled(),

172

help="Check if client is enabled")

173

parser.add_argument("-r", "--remove", dest="commands",

174

action="append_const", default=[],

175

const=command.Remove(),

176

help="Remove client")

177

parser.add_argument("-c", "--checker", dest="commands",

178

action="append", default=[],

179

metavar="COMMAND", type=command.SetChecker,

180

help="Set checker command for client")

181

parser.add_argument(

182

"-t", "--timeout", dest="commands", action="append",

183

default=[], metavar="TIME",

184

type=command.SetTimeout.argparse(string_to_delta),

185

help="Set timeout for client")

186

parser.add_argument(

187

"--extended-timeout", dest="commands", action="append",

188

default=[], metavar="TIME",

189

type=command.SetExtendedTimeout.argparse(string_to_delta),

190

help="Set extended timeout for client")

191

parser.add_argument(

192

"-i", "--interval", dest="commands", action="append",

193

default=[], metavar="TIME",

194

type=command.SetInterval.argparse(string_to_delta),

195

help="Set checker interval for client")

196

approve_deny_default = parser.add_mutually_exclusive_group()

197

approve_deny_default.add_argument(

198

"--approve-by-default", dest="commands",

199

action="append_const", default=[],

200

const=command.ApproveByDefault(),

201

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

202

approve_deny_default.add_argument(

203

"--deny-by-default", dest="commands",

204

action="append_const", default=[],

205

const=command.DenyByDefault(),

206

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

207

parser.add_argument(

208

"--approval-delay", dest="commands", action="append",

209

default=[], metavar="TIME",

210

type=command.SetApprovalDelay.argparse(string_to_delta),

211

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

212

parser.add_argument(

213

"--approval-duration", dest="commands", action="append",

214

default=[], metavar="TIME",

215

type=command.SetApprovalDuration.argparse(string_to_delta),

216

help="Set duration of one client approval")

217

parser.add_argument("-H", "--host", dest="commands",

218

action="append", default=[], metavar="STRING",

219

type=command.SetHost,

220

help="Set host for client")

221

parser.add_argument(

222

"-s", "--secret", dest="commands", action="append",

223

default=[], metavar="FILENAME",

224

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

225

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

226

approve_deny = parser.add_mutually_exclusive_group()

227

approve_deny.add_argument(

228

"-A", "--approve", dest="commands", action="append_const",

229

default=[], const=command.Approve(),

230

help="Approve any current client request")

231

approve_deny.add_argument("-D", "--deny", dest="commands",

232

action="append_const", default=[],

233

const=command.Deny(),

234

help="Deny any current client request")

235

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

236

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

237

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

238

help="Run self-test")

239

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

54

240

55

241

56

242

def string_to_delta(interval):

57

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

58

59

>>> string_to_delta('7d')

60

datetime.timedelta(7)

61

>>> string_to_delta('60s')

62

datetime.timedelta(0, 60)

63

>>> string_to_delta('60m')

64

datetime.timedelta(0, 3600)

65

>>> string_to_delta('24h')

66

datetime.timedelta(1)

67

>>> string_to_delta(u'1w')

68

datetime.timedelta(7)

69

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

70

datetime.timedelta(0, 330)

71

"""

72

timevalue = datetime.timedelta(0)

73

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

74

75

for s in regexp.findall(interval):

76

try:

77

suffix = unicode(s[-1])

78

value = int(s[:-1])

79

if suffix == u"d":

80

delta = datetime.timedelta(value)

81

elif suffix == u"s":

82

delta = datetime.timedelta(0, value)

83

elif suffix == u"m":

84

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

85

elif suffix == u"h":

86

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

87

elif suffix == u"w":

88

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

89

else:

90

raise ValueError

91

except (ValueError, IndexError):

92

raise ValueError

93

timevalue += delta

94

return timevalue

95

96

def print_clients(clients):

97

def valuetostring(value, keyword):

98

if type(value) is dbus.Boolean:

99

return u"Yes" if value else u"No"

100

if keyword in (u"timeout", u"interval"):

101

return milliseconds_to_string(value)

102

return unicode(value)

103

104

# Create format string to print table rows

105

format_string = u' '.join(u'%%-%ds' %

106

max(len(tablewords[key]),

107

max(len(valuetostring(client[key], key))

108

for client in

109

clients))

110

for key in keywords)

111

# Print header line

112

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

113

for client in clients:

114

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

115

for key in keywords)

116

117

parser = OptionParser(version = "%%prog %s" % version)

118

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

119

help="Print all fields")

120

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

121

help="Enable client")

122

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

123

help="disable client")

124

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

125

help="Bump timeout for client")

126

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

127

help="Start checker for client")

128

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

129

help="Stop checker for client")

130

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

131

help="Check if client is enabled")

132

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

133

help="Remove client")

134

parser.add_option("-c", "--checker", type="string",

135

help="Set checker command for client")

136

parser.add_option("-t", "--timeout", type="string",

137

help="Set timeout for client")

138

parser.add_option("-i", "--interval", type="string",

139

help="Set checker interval for client")

140

parser.add_option("-H", "--host", type="string",

141

help="Set host for client")

142

parser.add_option("-s", "--secret", type="string",

143

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

144

options, client_names = parser.parse_args()

145

146

# Compile list of clients to process

147

clients=[]

148

for name in client_names:

149

for path, client in mandos_clients.iteritems():

150

if client['name'] == name:

151

client_objc = bus.get_object(busname, path)

152

clients.append(client_objc)

153

break

154

else:

155

print >> sys.stderr, "Client not found on server: %r" % name

156

sys.exit(1)

157

158

if not clients and mandos_clients.values():

159

keywords = defaultkeywords

160

if options.all:

161

keywords = ('name', 'enabled', 'timeout', 'last_checked_ok',

162

'created', 'interval', 'host', 'fingerprint',

163

'checker_running', 'last_enabled', 'checker')

164

print_clients(mandos_clients.values())

165

166

# Process each client in the list by all selected options

167

for client in clients:

168

if options.remove:

169

mandos_serv.RemoveClient(client.__dbus_object_path__)

170

if options.enable:

171

client.Enable(dbus_interface=client_interface)

172

if options.disable:

173

client.Disable(dbus_interface=client_interface)

174

if options.bump_timeout:

175

client.CheckedOK(dbus_interface=client_interface)

176

if options.start_checker:

177

client.StartChecker(dbus_interface=client_interface)

178

if options.stop_checker:

179

client.StopChecker(dbus_interface=client_interface)

180

if options.is_enabled:

181

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

182

u"enabled",

183

dbus_interface=dbus.PROPERTIES_IFACE)

184

else 1)

185

if options.checker:

186

client.Set(client_interface, u"checker", options.checker,

187

dbus_interface=dbus.PROPERTIES_IFACE)

188

if options.host:

189

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

190

dbus_interface=dbus.PROPERTIES_IFACE)

191

if options.interval:

192

client.Set(client_interface, u"interval",

193

timedelta_to_milliseconds

194

(string_to_delta(options.interval)),

195

dbus_interface=dbus.PROPERTIES_IFACE)

196

if options.timeout:

197

client.Set(client_interface, u"timeout",

198

timedelta_to_milliseconds(string_to_delta

199

(options.timeout)),

200

dbus_interface=dbus.PROPERTIES_IFACE)

201

if options.secret:

202

client.Set(client_interface, u"secret",

203

dbus.ByteArray(open(options.secret, u'rb').read()),

204

dbus_interface=dbus.PROPERTIES_IFACE)

243

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

244

245

try:

246

return rfc3339_duration_to_delta(interval)

247

except ValueError as e:

248

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

249

' '.join(e.args))

250

return parse_pre_1_6_1_interval(interval)

251

252

253

def rfc3339_duration_to_delta(duration):

254

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

255

256

>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

257

True

258

>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

259

True

260

>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)

261

True

262

>>> # 60 months

263

>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)

264

True

265

>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

266

True

267

>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

268

True

269

>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

270

True

271

>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

272

True

273

>>> # Can not be empty:

274

>>> rfc3339_duration_to_delta("")

275

Traceback (most recent call last):

276

...

277

ValueError: Invalid RFC 3339 duration: ""

278

>>> # Must start with "P":

279

>>> rfc3339_duration_to_delta("1D")

280

Traceback (most recent call last):

281

...

282

ValueError: Invalid RFC 3339 duration: "1D"

283

>>> # Must use correct order

284

>>> rfc3339_duration_to_delta("PT1S2M")

285

Traceback (most recent call last):

286

...

287

ValueError: Invalid RFC 3339 duration: "PT1S2M"

288

>>> # Time needs time marker

289

>>> rfc3339_duration_to_delta("P1H2S")

290

Traceback (most recent call last):

291

...

292

ValueError: Invalid RFC 3339 duration: "P1H2S"

293

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

294

>>> rfc3339_duration_to_delta("P1D2W")

295

Traceback (most recent call last):

296

...

297

ValueError: Invalid RFC 3339 duration: "P1D2W"

298

>>> rfc3339_duration_to_delta("P2W2H")

299

Traceback (most recent call last):

300

...

301

ValueError: Invalid RFC 3339 duration: "P2W2H"

302

"""

303

304

# Parsing an RFC 3339 duration with regular expressions is not

305

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

306

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

307

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

308

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

309

# avoid excessive use of external libraries.

310

311

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

312

Token = collections.namedtuple("Token", (

313

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

314

# a "group" containing digits

315

"value", # datetime.timedelta or None

316

"followers")) # Tokens valid after this token

317

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

318

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

319

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

320

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

321

datetime.timedelta(seconds=1),

322

frozenset((token_end, )))

323

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

324

datetime.timedelta(minutes=1),

325

frozenset((token_second, token_end)))

326

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

327

datetime.timedelta(hours=1),

328

frozenset((token_minute, token_end)))

329

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

330

None,

331

frozenset((token_hour, token_minute,

332

token_second)))

333

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

334

datetime.timedelta(days=1),

335

frozenset((token_time, token_end)))

336

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

337

datetime.timedelta(weeks=4),

338

frozenset((token_day, token_end)))

339

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

340

datetime.timedelta(weeks=52),

341

frozenset((token_month, token_end)))

342

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

343

datetime.timedelta(weeks=1),

344

frozenset((token_end, )))

345

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

346

frozenset((token_year, token_month,

347

token_day, token_time,

348

token_week)))

349

# Define starting values:

350

# Value so far

351

value = datetime.timedelta()

352

found_token = None

353

# Following valid tokens

354

followers = frozenset((token_duration, ))

355

# String left to parse

356

s = duration

357

# Loop until end token is found

358

while found_token is not token_end:

359

# Search for any currently valid tokens

360

for token in followers:

361

match = token.regexp.match(s)

362

if match is not None:

363

# Token found

364

if token.value is not None:

365

# Value found, parse digits

366

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

367

# Add to value so far

368

value += factor * token.value

369

# Strip token from string

370

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

371

# Go to found token

372

found_token = token

373

# Set valid next tokens

374

followers = found_token.followers

375

break

376

else:

377

# No currently valid tokens were found

378

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

379

.format(duration))

380

# End token found

381

return value

382

383

384

def parse_pre_1_6_1_interval(interval):

385

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

386

and return a datetime.timedelta

387

388

>>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)

389

True

390

>>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)

391

True

392

>>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)

393

True

394

>>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)

395

True

396

>>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)

397

True

398

>>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)

399

True

400

>>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)

401

True

402

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

403

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)

404

True

405

406

"""

407

408

value = datetime.timedelta(0)

409

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

410

411

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

412

if suffix == "d":

413

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

414

elif suffix == "s":

415

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

416

elif suffix == "m":

417

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

418

elif suffix == "h":

419

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

420

elif suffix == "w":

421

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

422

elif suffix == "":

423

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

424

return value

425

426

427

def check_option_syntax(parser, options):

428

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

429

argparse"""

430

431

def has_commands(options, commands=None):

432

if commands is None:

433

commands = (command.Enable,

434

command.Disable,

435

command.BumpTimeout,

436

command.StartChecker,

437

command.StopChecker,

438

command.IsEnabled,

439

command.Remove,

440

command.SetChecker,

441

command.SetTimeout,

442

command.SetExtendedTimeout,

443

command.SetInterval,

444

command.ApproveByDefault,

445

command.DenyByDefault,

446

command.SetApprovalDelay,

447

command.SetApprovalDuration,

448

command.SetHost,

449

command.SetSecret,

450

command.Approve,

451

command.Deny)

452

return any(isinstance(cmd, commands)

453

for cmd in options.commands)

454

455

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

456

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

457

if options.verbose and has_commands(options):

458

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

459

if (has_commands(options, (command.DumpJSON,))

460

and (options.verbose or len(options.commands) > 1)):

461

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

462

if options.all and not has_commands(options):

463

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

464

if (has_commands(options, (command.IsEnabled,))

465

and len(options.client) > 1):

466

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

467

if (len(options.commands) > 1

468

and has_commands(options, (command.Remove,))

469

and not has_commands(options, (command.Deny,))):

470

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

471

472

473

class dbus:

474

475

class SystemBus:

476

477

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

478

def get_managed_objects(self, busname, objectpath):

479

return self.call_method("GetManagedObjects", busname,

480

objectpath,

481

self.object_manager_iface)

482

483

properties_iface = "org.freedesktop.DBus.Properties"

484

def set_property(self, busname, objectpath, interface, key,

485

value):

486

self.call_method("Set", busname, objectpath,

487

self.properties_iface, interface, key,

488

value)

489

490

491

class MandosBus(SystemBus):

492

busname_domain = "se.recompile"

493

busname = busname_domain + ".Mandos"

494

server_path = "/"

495

server_interface = busname_domain + ".Mandos"

496

client_interface = busname_domain + ".Mandos.Client"

497

del busname_domain

498

499

def get_clients_and_properties(self):

500

managed_objects = self.get_managed_objects(

501

self.busname, self.server_path)

502

return {objpath: properties[self.client_interface]

503

for objpath, properties in managed_objects.items()

504

if self.client_interface in properties}

505

506

def set_client_property(self, objectpath, key, value):

507

return self.set_property(self.busname, objectpath,

508

self.client_interface, key,

509

value)

510

511

def call_client_method(self, objectpath, method, *args):

512

return self.call_method(method, self.busname, objectpath,

513

self.client_interface, *args)

514

515

def call_server_method(self, method, *args):

516

return self.call_method(method, self.busname,

517

self.server_path,

518

self.server_interface, *args)

519

520

class Error(Exception):

521

pass

522

523

class ConnectFailed(Error):

524

pass

525

526

527

class dbus_python_adapter:

528

529

class SystemBus(dbus.MandosBus):

530

"""Use dbus-python"""

531

532

def __init__(self, module=dbus_python):

533

self.dbus_python = module

534

self.bus = self.dbus_python.SystemBus()

535

536

@contextlib.contextmanager

537

def convert_exception(self, exception_class=dbus.Error):

538

try:

539

yield

540

except self.dbus_python.exceptions.DBusException as e:

541

# This does what "raise from" would do

542

exc = exception_class(*e.args)

543

exc.__cause__ = e

544

raise exc

545

546

def call_method(self, methodname, busname, objectpath,

547

interface, *args):

548

proxy_object = self.get_object(busname, objectpath)

549

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

550

interface, methodname,

551

", ".join(repr(a) for a in args))

552

method = getattr(proxy_object, methodname)

553

with self.convert_exception():

554

with dbus_python_adapter.SilenceLogger(

555

"dbus.proxies"):

556

value = method(*args, dbus_interface=interface)

557

return self.type_filter(value)

558

559

def get_object(self, busname, objectpath):

560

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

561

busname, objectpath)

562

with self.convert_exception(dbus.ConnectFailed):

563

return self.bus.get_object(busname, objectpath)

564

565

def type_filter(self, value):

566

"""Convert the most bothersome types to Python types"""

567

if isinstance(value, self.dbus_python.Boolean):

568

return bool(value)

569

if isinstance(value, self.dbus_python.ObjectPath):

570

return str(value)

571

# Also recurse into dictionaries

572

if isinstance(value, self.dbus_python.Dictionary):

573

return {self.type_filter(key):

574

self.type_filter(subval)

575

for key, subval in value.items()}

576

return value

577

578

def set_client_property(self, objectpath, key, value):

579

if key == "Secret":

580

if not isinstance(value, bytes):

581

value = value.encode("utf-8")

582

value = self.dbus_python.ByteArray(value)

583

return self.set_property(self.busname, objectpath,

584

self.client_interface, key,

585

value)

586

587

class SilenceLogger:

588

"Simple context manager to silence a particular logger"

589

def __init__(self, loggername):

590

self.logger = logging.getLogger(loggername)

591

592

def __enter__(self):

593

self.logger.addFilter(self.nullfilter)

594

595

class NullFilter(logging.Filter):

596

def filter(self, record):

597

return False

598

599

nullfilter = NullFilter()

600

601

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

602

self.logger.removeFilter(self.nullfilter)

603

604

605

class CachingBus(SystemBus):

606

"""A caching layer for dbus_python_adapter.SystemBus"""

607

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

608

self.object_cache = {}

609

super(dbus_python_adapter.CachingBus,

610

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

611

def get_object(self, busname, objectpath):

612

try:

613

return self.object_cache[(busname, objectpath)]

614

except KeyError:

615

new_object = super(

616

dbus_python_adapter.CachingBus,

617

self).get_object(busname, objectpath)

618

self.object_cache[(busname, objectpath)] = new_object

619

return new_object

620

621

622

class pydbus_adapter:

623

class SystemBus(dbus.MandosBus):

624

def __init__(self, module=pydbus):

625

self.pydbus = module

626

self.bus = self.pydbus.SystemBus()

627

628

@contextlib.contextmanager

629

def convert_exception(self, exception_class=dbus.Error):

630

try:

631

yield

632

except gi.repository.GLib.Error as e:

633

# This does what "raise from" would do

634

exc = exception_class(*e.args)

635

exc.__cause__ = e

636

raise exc

637

638

def call_method(self, methodname, busname, objectpath,

639

interface, *args):

640

proxy_object = self.get(busname, objectpath)

641

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

642

interface, methodname,

643

", ".join(repr(a) for a in args))

644

method = getattr(proxy_object[interface], methodname)

645

with self.convert_exception():

646

return method(*args)

647

648

def get(self, busname, objectpath):

649

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

650

busname, objectpath)

651

with self.convert_exception(dbus.ConnectFailed):

652

if sys.version_info.major <= 2:

653

with warnings.catch_warnings():

654

warnings.filterwarnings(

655

"ignore", "", DeprecationWarning,

656

r"^xml\.etree\.ElementTree$")

657

return self.bus.get(busname, objectpath)

658

else:

659

return self.bus.get(busname, objectpath)

660

661

def set_property(self, busname, objectpath, interface, key,

662

value):

663

proxy_object = self.get(busname, objectpath)

664

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

665

objectpath, self.properties_iface, interface,

666

key, value)

667

setattr(proxy_object[interface], key, value)

668

669

class CachingBus(SystemBus):

670

"""A caching layer for pydbus_adapter.SystemBus"""

671

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

672

self.object_cache = {}

673

super(pydbus_adapter.CachingBus,

674

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

675

def get(self, busname, objectpath):

676

try:

677

return self.object_cache[(busname, objectpath)]

678

except KeyError:

679

new_object = (super(pydbus_adapter.CachingBus, self)

680

.get(busname, objectpath))

681

self.object_cache[(busname, objectpath)] = new_object

682

return new_object

683

684

685

def commands_from_options(options):

686

687

commands = list(options.commands)

688

689

def find_cmd(cmd, commands):

690

i = 0

691

for i, c in enumerate(commands):

692

if isinstance(c, cmd):

693

return i

694

return i+1

695

696

# If command.Remove is present, move any instances of command.Deny

697

# to occur ahead of command.Remove.

698

index_of_remove = find_cmd(command.Remove, commands)

699

before_remove = commands[:index_of_remove]

700

after_remove = commands[index_of_remove:]

701

cleaned_after = []

702

for cmd in after_remove:

703

if isinstance(cmd, command.Deny):

704

before_remove.append(cmd)

705

else:

706

cleaned_after.append(cmd)

707

if cleaned_after != after_remove:

708

commands = before_remove + cleaned_after

709

710

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

711

# optionally verbosely

712

if not commands:

713

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

714

715

return commands

716

717

718

class command:

719

"""A namespace for command classes"""

720

721

class Base:

722

"""Abstract base class for commands"""

723

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

724

"""Normal commands should implement run_on_one_client(),

725

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

726

override this run() method instead.

727

"""

728

self.bus = bus

729

for client, properties in clients.items():

730

self.run_on_one_client(client, properties)

731

732

733

class IsEnabled(Base):

734

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

735

properties = next(iter(clients.values()))

736

if properties["Enabled"]:

737

sys.exit(0)

738

sys.exit(1)

739

740

741

class Approve(Base):

742

def run_on_one_client(self, client, properties):

743

self.bus.call_client_method(client, "Approve", True)

744

745

746

class Deny(Base):

747

def run_on_one_client(self, client, properties):

748

self.bus.call_client_method(client, "Approve", False)

749

750

751

class Remove(Base):

752

def run(self, clients, bus):

753

for clientpath in frozenset(clients.keys()):

754

bus.call_server_method("RemoveClient", clientpath)

755

756

757

class Output(Base):

758

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

759

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

760

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

761

"Fingerprint", "CheckerRunning",

762

"LastEnabled", "ApprovalPending",

763

"ApprovedByDefault", "LastApprovalRequest",

764

"ApprovalDelay", "ApprovalDuration",

765

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

766

"LastCheckerStatus")

767

768

769

class DumpJSON(Output):

770

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

771

data = {properties["Name"]:

772

{key: properties[key]

773

for key in self.all_keywords}

774

for properties in clients.values()}

775

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

776

777

778

class PrintTable(Output):

779

def __init__(self, verbose=False):

780

self.verbose = verbose

781

782

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

783

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

784

"LastCheckedOK")

785

keywords = default_keywords

786

if self.verbose:

787

keywords = self.all_keywords

788

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

789

790

class TableOfClients:

791

tableheaders = {

792

"Name": "Name",

793

"Enabled": "Enabled",

794

"Timeout": "Timeout",

795

"LastCheckedOK": "Last Successful Check",

796

"LastApprovalRequest": "Last Approval Request",

797

"Created": "Created",

798

"Interval": "Interval",

799

"Host": "Host",

800

"Fingerprint": "Fingerprint",

801

"KeyID": "Key ID",

802

"CheckerRunning": "Check Is Running",

803

"LastEnabled": "Last Enabled",

804

"ApprovalPending": "Approval Is Pending",

805

"ApprovedByDefault": "Approved By Default",

806

"ApprovalDelay": "Approval Delay",

807

"ApprovalDuration": "Approval Duration",

808

"Checker": "Checker",

809

"ExtendedTimeout": "Extended Timeout",

810

"Expires": "Expires",

811

"LastCheckerStatus": "Last Checker Status",

812

}

813

814

def __init__(self, clients, keywords):

815

self.clients = clients

816

self.keywords = keywords

817

818

def __str__(self):

819

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

820

821

if sys.version_info.major == 2:

822

__unicode__ = __str__

823

def __str__(self):

824

return str(self).encode(

825

locale.getpreferredencoding())

826

827

def rows(self):

828

format_string = self.row_formatting_string()

829

rows = [self.header_line(format_string)]

830

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

831

for client in self.clients)

832

return rows

833

834

def row_formatting_string(self):

835

"Format string used to format table rows"

836

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

837

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

838

*(len(self.string_from_client(client,

839

key))

840

for client in self.clients)),

841

key=key)

842

for key in self.keywords)

843

844

def string_from_client(self, client, key):

845

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

846

847

@classmethod

848

def valuetostring(cls, value, keyword):

849

if isinstance(value, bool):

850

return "Yes" if value else "No"

851

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

852

"ApprovalDuration", "ExtendedTimeout"):

853

return cls.milliseconds_to_string(value)

854

return str(value)

855

856

def header_line(self, format_string):

857

return format_string.format(**self.tableheaders)

858

859

def client_line(self, client, format_string):

860

return format_string.format(

861

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

862

for key in self.keywords})

863

864

@staticmethod

865

def milliseconds_to_string(ms):

866

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

867

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

868

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

869

if td.days else "",

870

hours=td.seconds // 3600,

871

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

872

seconds=td.seconds % 60))

873

874

875

class PropertySetter(Base):

876

"Abstract class for Actions for setting one client property"

877

878

def run_on_one_client(self, client, properties=None):

879

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

880

self.bus.set_client_property(client, self.propname,

881

self.value_to_set)

882

883

@property

884

def propname(self):

885

raise NotImplementedError()

886

887

888

class Enable(PropertySetter):

889

propname = "Enabled"

890

value_to_set = True

891

892

893

class Disable(PropertySetter):

894

propname = "Enabled"

895

value_to_set = False

896

897

898

class BumpTimeout(PropertySetter):

899

propname = "LastCheckedOK"

900

value_to_set = ""

901

902

903

class StartChecker(PropertySetter):

904

propname = "CheckerRunning"

905

value_to_set = True

906

907

908

class StopChecker(PropertySetter):

909

propname = "CheckerRunning"

910

value_to_set = False

911

912

913

class ApproveByDefault(PropertySetter):

914

propname = "ApprovedByDefault"

915

value_to_set = True

916

917

918

class DenyByDefault(PropertySetter):

919

propname = "ApprovedByDefault"

920

value_to_set = False

921

922

923

class PropertySetterValue(PropertySetter):

924

"""Abstract class for PropertySetter recieving a value as

925

constructor argument instead of a class attribute."""

926

def __init__(self, value):

927

self.value_to_set = value

928

929

@classmethod

930

def argparse(cls, argtype):

931

def cmdtype(arg):

932

return cls(argtype(arg))

933

return cmdtype

934

935

class SetChecker(PropertySetterValue):

936

propname = "Checker"

937

938

939

class SetHost(PropertySetterValue):

940

propname = "Host"

941

942

943

class SetSecret(PropertySetterValue):

944

propname = "Secret"

945

946

@property

947

def value_to_set(self):

948

return self._vts

949

950

@value_to_set.setter

951

def value_to_set(self, value):

952

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

953

self._vts = value.read()

954

value.close()

955

956

957

class PropertySetterValueMilliseconds(PropertySetterValue):

958

"""Abstract class for PropertySetterValue taking a value

959

argument as a datetime.timedelta() but should store it as

960

milliseconds."""

961

962

@property

963

def value_to_set(self):

964

return self._vts

965

966

@value_to_set.setter

967

def value_to_set(self, value):

968

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

969

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

970

971

972

class SetTimeout(PropertySetterValueMilliseconds):

973

propname = "Timeout"

974

975

976

class SetExtendedTimeout(PropertySetterValueMilliseconds):

977

propname = "ExtendedTimeout"

978

979

980

class SetInterval(PropertySetterValueMilliseconds):

981

propname = "Interval"

982

983

984

class SetApprovalDelay(PropertySetterValueMilliseconds):

985

propname = "ApprovalDelay"

986

987

988

class SetApprovalDuration(PropertySetterValueMilliseconds):

989

propname = "ApprovalDuration"

990

991

992

993

class TestCaseWithAssertLogs(unittest.TestCase):

994

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

995

996

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

997

@contextlib.contextmanager

998

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

999

capturing_handler = self.CapturingLevelHandler(level)

1000

old_level = logger.level

1001

old_propagate = logger.propagate

1002

logger.addHandler(capturing_handler)

1003

logger.setLevel(level)

1004

logger.propagate = False

1005

try:

1006

yield capturing_handler.watcher

1007

finally:

1008

logger.propagate = old_propagate

1009

logger.removeHandler(capturing_handler)

1010

logger.setLevel(old_level)

1011

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

1012

0)

1013

1014

class CapturingLevelHandler(logging.Handler):

1015

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

1016

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

1017

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

1018

def emit(self, record):

1019

self.watcher.records.append(record)

1020

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

1021

1022

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1023

("records",

1024

"output"))

1025

1026

1027

class Unique:

1028

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

1029

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

1030

1031

1032

class Test_string_to_delta(TestCaseWithAssertLogs):

1033

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

1034

# rfc3339_duration_to_delta() already has more comprehensive

1035

# tests, which are run by doctest.

1036

1037

def test_rfc3339_zero_seconds(self):

1038

self.assertEqual(datetime.timedelta(),

1039

string_to_delta("PT0S"))

1040

1041

def test_rfc3339_zero_days(self):

1042

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

1043

1044

def test_rfc3339_one_second(self):

1045

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

1046

string_to_delta("PT1S"))

1047

1048

def test_rfc3339_two_hours(self):

1049

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

1050

string_to_delta("PT2H"))

1051

1052

def test_falls_back_to_pre_1_6_1_with_warning(self):

1053

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

1054

value = string_to_delta("2h")

1055

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

1056

1057

1058

class Test_check_option_syntax(unittest.TestCase):

1059

def setUp(self):

1060

self.parser = argparse.ArgumentParser()

1061

add_command_line_options(self.parser)

1062

1063

def test_actions_requires_client_or_all(self):

1064

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

1065

args = self.actionargs(action, value)

1066

with self.assertParseError():

1067

self.parse_args(args)

1068

1069

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

1070

# check_option_syntax()

1071

actions = {

1072

"--enable": None,

1073

"--disable": None,

1074

"--bump-timeout": None,

1075

"--start-checker": None,

1076

"--stop-checker": None,

1077

"--is-enabled": None,

1078

"--remove": None,

1079

"--checker": "x",

1080

"--timeout": "PT0S",

1081

"--extended-timeout": "PT0S",

1082

"--interval": "PT0S",

1083

"--approve-by-default": None,

1084

"--deny-by-default": None,

1085

"--approval-delay": "PT0S",

1086

"--approval-duration": "PT0S",

1087

"--host": "hostname",

1088

"--secret": "/dev/null",

1089

"--approve": None,

1090

"--deny": None,

1091

}

1092

1093

@staticmethod

1094

def actionargs(action, value, *args):

1095

if value is not None:

1096

return [action, value] + list(args)

1097

else:

1098

return [action] + list(args)

1099

1100

@contextlib.contextmanager

1101

def assertParseError(self):

1102

with self.assertRaises(SystemExit) as e:

1103

with self.redirect_stderr_to_devnull():

1104

yield

1105

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

1106

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

1107

# /argparse.html#exiting-methods

1108

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

1109

1110

def parse_args(self, args):

1111

options = self.parser.parse_args(args)

1112

check_option_syntax(self.parser, options)

1113

1114

@staticmethod

1115

@contextlib.contextmanager

1116

def redirect_stderr_to_devnull():

1117

old_stderr = sys.stderr

1118

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

1119

sys.stderr = null

1120

try:

1121

yield

1122

finally:

1123

sys.stderr = old_stderr

1124

1125

def check_option_syntax(self, options):

1126

check_option_syntax(self.parser, options)

1127

1128

def test_actions_all_conflicts_with_verbose(self):

1129

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

1130

args = self.actionargs(action, value, "--all",

1131

"--verbose")

1132

with self.assertParseError():

1133

self.parse_args(args)

1134

1135

def test_actions_with_client_conflicts_with_verbose(self):

1136

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

1137

args = self.actionargs(action, value, "--verbose",

1138

"client")

1139

with self.assertParseError():

1140

self.parse_args(args)

1141

1142

def test_dump_json_conflicts_with_verbose(self):

1143

args = ["--dump-json", "--verbose"]

1144

with self.assertParseError():

1145

self.parse_args(args)

1146

1147

def test_dump_json_conflicts_with_action(self):

1148

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

1149

args = self.actionargs(action, value, "--dump-json")

1150

with self.assertParseError():

1151

self.parse_args(args)

1152

1153

def test_all_can_not_be_alone(self):

1154

args = ["--all"]

1155

with self.assertParseError():

1156

self.parse_args(args)

1157

1158

def test_all_is_ok_with_any_action(self):

1159

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

1160

args = self.actionargs(action, value, "--all")

1161

self.parse_args(args)

1162

1163

def test_any_action_is_ok_with_one_client(self):

1164

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

1165

args = self.actionargs(action, value, "client")

1166

self.parse_args(args)

1167

1168

def test_one_client_with_all_actions_except_is_enabled(self):

1169

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

1170

if action == "--is-enabled":

1171

continue

1172

args = self.actionargs(action, value, "client")

1173

self.parse_args(args)

1174

1175

def test_two_clients_with_all_actions_except_is_enabled(self):

1176

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

1177

if action == "--is-enabled":

1178

continue

1179

args = self.actionargs(action, value, "client1",

1180

"client2")

1181

self.parse_args(args)

1182

1183

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1184

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

1185

if action == "--is-enabled":

1186

continue

1187

args = self.actionargs(action, value, "client1",

1188

"client2")

1189

self.parse_args(args)

1190

1191

def test_is_enabled_fails_without_client(self):

1192

args = ["--is-enabled"]

1193

with self.assertParseError():

1194

self.parse_args(args)

1195

1196

def test_is_enabled_fails_with_two_clients(self):

1197

args = ["--is-enabled", "client1", "client2"]

1198

with self.assertParseError():

1199

self.parse_args(args)

1200

1201

def test_remove_can_only_be_combined_with_action_deny(self):

1202

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

1203

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

1204

continue

1205

args = self.actionargs(action, value, "--all",

1206

"--remove")

1207

with self.assertParseError():

1208

self.parse_args(args)

1209

1210

1211

class Test_dbus_exceptions(unittest.TestCase):

1212

1213

def test_dbus_ConnectFailed_is_Error(self):

1214

with self.assertRaises(dbus.Error):

1215

raise dbus.ConnectFailed()

1216

1217

1218

class Test_dbus_MandosBus(unittest.TestCase):

1219

1220

class MockMandosBus(dbus.MandosBus):

1221

def __init__(self):

1222

self._name = "se.recompile.Mandos"

1223

self._server_path = "/"

1224

self._server_interface = "se.recompile.Mandos"

1225

self._client_interface = "se.recompile.Mandos.Client"

1226

self.calls = []

1227

self.call_method_return = Unique()

1228

1229

def call_method(self, methodname, busname, objectpath,

1230

interface, *args):

1231

self.calls.append((methodname, busname, objectpath,

1232

interface, args))

1233

return self.call_method_return

1234

1235

def setUp(self):

1236

self.bus = self.MockMandosBus()

1237

1238

def test_set_client_property(self):

1239

self.bus.set_client_property("objectpath", "key", "value")

1240

expected_call = ("Set", self.bus._name, "objectpath",

1241

"org.freedesktop.DBus.Properties",

1242

(self.bus._client_interface, "key", "value"))

1243

self.assertIn(expected_call, self.bus.calls)

1244

1245

def test_call_client_method(self):

1246

ret = self.bus.call_client_method("objectpath", "methodname")

1247

self.assertIs(self.bus.call_method_return, ret)

1248

expected_call = ("methodname", self.bus._name, "objectpath",

1249

self.bus._client_interface, ())

1250

self.assertIn(expected_call, self.bus.calls)

1251

1252

def test_call_client_method_with_args(self):

1253

args = (Unique(), Unique())

1254

ret = self.bus.call_client_method("objectpath", "methodname",

1255

*args)

1256

self.assertIs(self.bus.call_method_return, ret)

1257

expected_call = ("methodname", self.bus._name, "objectpath",

1258

self.bus._client_interface,

1259

(args[0], args[1]))

1260

self.assertIn(expected_call, self.bus.calls)

1261

1262

def test_get_clients_and_properties(self):

1263

managed_objects = {

1264

"objectpath": {

1265

self.bus._client_interface: {

1266

"key": "value",

1267

"bool": True,

1268

},

1269

"irrelevant_interface": {

1270

"key": "othervalue",

1271

"bool": False,

1272

},

1273

},

1274

"other_objectpath": {

1275

"other_irrelevant_interface": {

1276

"key": "value 3",

1277

"bool": None,

1278

},

1279

},

1280

}

1281

expected_clients_and_properties = {

1282

"objectpath": {

1283

"key": "value",

1284

"bool": True,

1285

}

1286

}

1287

self.bus.call_method_return = managed_objects

1288

ret = self.bus.get_clients_and_properties()

1289

self.assertDictEqual(expected_clients_and_properties, ret)

1290

expected_call = ("GetManagedObjects", self.bus._name,

1291

self.bus._server_path,

1292

"org.freedesktop.DBus.ObjectManager", ())

1293

self.assertIn(expected_call, self.bus.calls)

1294

1295

def test_call_server_method(self):

1296

ret = self.bus.call_server_method("methodname")

1297

self.assertIs(self.bus.call_method_return, ret)

1298

expected_call = ("methodname", self.bus._name,

1299

self.bus._server_path,

1300

self.bus._server_interface, ())

1301

self.assertIn(expected_call, self.bus.calls)

1302

1303

def test_call_server_method_with_args(self):

1304

args = (Unique(), Unique())

1305

ret = self.bus.call_server_method("methodname", *args)

1306

self.assertIs(self.bus.call_method_return, ret)

1307

expected_call = ("methodname", self.bus._name,

1308

self.bus._server_path,

1309

self.bus._server_interface,

1310

(args[0], args[1]))

1311

self.assertIn(expected_call, self.bus.calls)

1312

1313

1314

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1315

1316

def MockDBusPython_func(self, func):

1317

class mock_dbus_python:

1318

"""mock dbus-python module"""

1319

class exceptions:

1320

"""Pseudo-namespace"""

1321

class DBusException(Exception):

1322

pass

1323

class SystemBus:

1324

@staticmethod

1325

def get_object(busname, objectpath):

1326

DBusObject = collections.namedtuple(

1327

"DBusObject", ("methodname", "Set"))

1328

def method(*args, **kwargs):

1329

self.assertEqual({"dbus_interface":

1330

"interface"},

1331

kwargs)

1332

return func(*args)

1333

def set_property(interface, key, value,

1334

dbus_interface=None):

1335

self.assertEqual(

1336

"org.freedesktop.DBus.Properties",

1337

dbus_interface)

1338

self.assertEqual("Secret", key)

1339

return func(interface, key, value,

1340

dbus_interface=dbus_interface)

1341

return DBusObject(methodname=method,

1342

Set=set_property)

1343

class Boolean:

1344

def __init__(self, value):

1345

self.value = bool(value)

1346

def __bool__(self):

1347

return self.value

1348

if sys.version_info.major == 2:

1349

__nonzero__ = __bool__

1350

class ObjectPath(str):

1351

pass

1352

class Dictionary(dict):

1353

pass

1354

class ByteArray(bytes):

1355

pass

1356

return mock_dbus_python

1357

1358

def call_method(self, bus, methodname, busname, objectpath,

1359

interface, *args):

1360

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

1361

return bus.call_method(methodname, busname, objectpath,

1362

interface, *args)

1363

1364

def test_call_method_returns(self):

1365

expected_method_return = Unique()

1366

method_args = (Unique(), Unique())

1367

def func(*args):

1368

self.assertEqual(len(method_args), len(args))

1369

for marg, arg in zip(method_args, args):

1370

self.assertIs(marg, arg)

1371

return expected_method_return

1372

mock_dbus_python = self.MockDBusPython_func(func)

1373

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1374

ret = self.call_method(bus, "methodname", "busname",

1375

"objectpath", "interface",

1376

*method_args)

1377

self.assertIs(ret, expected_method_return)

1378

1379

def test_call_method_filters_bool_true(self):

1380

def func():

1381

return method_return

1382

mock_dbus_python = self.MockDBusPython_func(func)

1383

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1384

method_return = mock_dbus_python.Boolean(True)

1385

ret = self.call_method(bus, "methodname", "busname",

1386

"objectpath", "interface")

1387

self.assertTrue(ret)

1388

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1389

1390

def test_call_method_filters_bool_false(self):

1391

def func():

1392

return method_return

1393

mock_dbus_python = self.MockDBusPython_func(func)

1394

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1395

method_return = mock_dbus_python.Boolean(False)

1396

ret = self.call_method(bus, "methodname", "busname",

1397

"objectpath", "interface")

1398

self.assertFalse(ret)

1399

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1400

1401

def test_call_method_filters_objectpath(self):

1402

def func():

1403

return method_return

1404

mock_dbus_python = self.MockDBusPython_func(func)

1405

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1406

method_return = mock_dbus_python.ObjectPath("objectpath")

1407

ret = self.call_method(bus, "methodname", "busname",

1408

"objectpath", "interface")

1409

self.assertEqual("objectpath", ret)

1410

self.assertIsNot("objectpath", ret)

1411

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1412

1413

def test_call_method_filters_booleans_in_dict(self):

1414

def func():

1415

return method_return

1416

mock_dbus_python = self.MockDBusPython_func(func)

1417

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1418

method_return = mock_dbus_python.Dictionary(

1419

{mock_dbus_python.Boolean(True):

1420

mock_dbus_python.Boolean(False),

1421

mock_dbus_python.Boolean(False):

1422

mock_dbus_python.Boolean(True)})

1423

ret = self.call_method(bus, "methodname", "busname",

1424

"objectpath", "interface")

1425

expected_method_return = {True: False,

1426

False: True}

1427

self.assertEqual(expected_method_return, ret)

1428

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1429

1430

def test_call_method_filters_objectpaths_in_dict(self):

1431

def func():

1432

return method_return

1433

mock_dbus_python = self.MockDBusPython_func(func)

1434

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1435

method_return = mock_dbus_python.Dictionary(

1436

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1437

mock_dbus_python.ObjectPath("objectpath_value_1"),

1438

mock_dbus_python.ObjectPath("objectpath_key_2"):

1439

mock_dbus_python.ObjectPath("objectpath_value_2")})

1440

ret = self.call_method(bus, "methodname", "busname",

1441

"objectpath", "interface")

1442

expected_method_return = {str(key): str(value)

1443

for key, value in

1444

method_return.items()}

1445

self.assertEqual(expected_method_return, ret)

1446

self.assertIsInstance(ret, dict)

1447

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1448

1449

def test_call_method_filters_dict_in_dict(self):

1450

def func():

1451

return method_return

1452

mock_dbus_python = self.MockDBusPython_func(func)

1453

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1454

method_return = mock_dbus_python.Dictionary(

1455

{"key1": mock_dbus_python.Dictionary({"key11": "value11",

1456

"key12": "value12"}),

1457

"key2": mock_dbus_python.Dictionary({"key21": "value21",

1458

"key22": "value22"})})

1459

ret = self.call_method(bus, "methodname", "busname",

1460

"objectpath", "interface")

1461

expected_method_return = {

1462

"key1": {"key11": "value11",

1463

"key12": "value12"},

1464

"key2": {"key21": "value21",

1465

"key22": "value22"},

1466

}

1467

self.assertEqual(expected_method_return, ret)

1468

self.assertIsInstance(ret, dict)

1469

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1470

for key, value in ret.items():

1471

self.assertIsInstance(value, dict)

1472

self.assertEqual(expected_method_return[key], value)

1473

self.assertNotIsInstance(value,

1474

mock_dbus_python.Dictionary)

1475

1476

def test_call_method_filters_dict_three_deep(self):

1477

def func():

1478

return method_return

1479

mock_dbus_python = self.MockDBusPython_func(func)

1480

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1481

method_return = mock_dbus_python.Dictionary(

1482

{"key1":

1483

mock_dbus_python.Dictionary(

1484

{"key2":

1485

mock_dbus_python.Dictionary(

1486

{"key3":

1487

mock_dbus_python.Boolean(True),

1488

}),

1489

}),

1490

})

1491

ret = self.call_method(bus, "methodname", "busname",

1492

"objectpath", "interface")

1493

expected_method_return = {"key1": {"key2": {"key3": True}}}

1494

self.assertEqual(expected_method_return, ret)

1495

self.assertIsInstance(ret, dict)

1496

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1497

self.assertIsInstance(ret["key1"], dict)

1498

self.assertNotIsInstance(ret["key1"],

1499

mock_dbus_python.Dictionary)

1500

self.assertIsInstance(ret["key1"]["key2"], dict)

1501

self.assertNotIsInstance(ret["key1"]["key2"],

1502

mock_dbus_python.Dictionary)

1503

self.assertTrue(ret["key1"]["key2"]["key3"])

1504

self.assertNotIsInstance(ret["key1"]["key2"]["key3"],

1505

mock_dbus_python.Boolean)

1506

1507

def test_call_method_handles_exception(self):

1508

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

1509

1510

def func():

1511

dbus_logger.error("Test")

1512

raise mock_dbus_python.exceptions.DBusException()

1513

1514

mock_dbus_python = self.MockDBusPython_func(func)

1515

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1516

1517

class CountingHandler(logging.Handler):

1518

count = 0

1519

def emit(self, record):

1520

self.count += 1

1521

1522

counting_handler = CountingHandler()

1523

1524

dbus_logger.addHandler(counting_handler)

1525

1526

try:

1527

with self.assertRaises(dbus.Error) as e:

1528

self.call_method(bus, "methodname", "busname",

1529

"objectpath", "interface")

1530

finally:

1531

dbus_logger.removeFilter(counting_handler)

1532

1533

self.assertNotIsInstance(e, dbus.ConnectFailed)

1534

1535

# Make sure the dbus logger was suppressed

1536

self.assertEqual(0, counting_handler.count)

1537

1538

def test_Set_Secret_sends_bytearray(self):

1539

ret = [None]

1540

def func(*args, **kwargs):

1541

ret[0] = (args, kwargs)

1542

mock_dbus_python = self.MockDBusPython_func(func)

1543

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1544

bus.set_client_property("objectpath", "Secret", "value")

1545

expected_call = (("se.recompile.Mandos.Client", "Secret",

1546

mock_dbus_python.ByteArray(b"value")),

1547

{"dbus_interface":

1548

"org.freedesktop.DBus.Properties"})

1549

self.assertEqual(expected_call, ret[0])

1550

if sys.version_info.major == 2:

1551

self.assertIsInstance(ret[0][0][-1],

1552

mock_dbus_python.ByteArray)

1553

1554

def test_get_object_converts_to_correct_exception(self):

1555

bus = dbus_python_adapter.SystemBus(

1556

self.fake_dbus_python_raises_exception_on_connect)

1557

with self.assertRaises(dbus.ConnectFailed):

1558

self.call_method(bus, "methodname", "busname",

1559

"objectpath", "interface")

1560

1561

class fake_dbus_python_raises_exception_on_connect:

1562

"""fake dbus-python module"""

1563

class exceptions:

1564

"""Pseudo-namespace"""

1565

class DBusException(Exception):

1566

pass

1567

1568

@classmethod

1569

def SystemBus(cls):

1570

def get_object(busname, objectpath):

1571

raise cls.exceptions.DBusException()

1572

Bus = collections.namedtuple("Bus", ["get_object"])

1573

return Bus(get_object=get_object)

1574

1575

1576

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1577

class mock_dbus_python:

1578

"""mock dbus-python modules"""

1579

class SystemBus:

1580

@staticmethod

1581

def get_object(busname, objectpath):

1582

return Unique()

1583

1584

def setUp(self):

1585

self.bus = dbus_python_adapter.CachingBus(

1586

self.mock_dbus_python)

1587

1588

def test_returns_distinct_objectpaths(self):

1589

obj1 = self.bus.get_object("busname", "objectpath1")

1590

self.assertIsInstance(obj1, Unique)

1591

obj2 = self.bus.get_object("busname", "objectpath2")

1592

self.assertIsInstance(obj2, Unique)

1593

self.assertIsNot(obj1, obj2)

1594

1595

def test_returns_distinct_busnames(self):

1596

obj1 = self.bus.get_object("busname1", "objectpath")

1597

self.assertIsInstance(obj1, Unique)

1598

obj2 = self.bus.get_object("busname2", "objectpath")

1599

self.assertIsInstance(obj2, Unique)

1600

self.assertIsNot(obj1, obj2)

1601

1602

def test_returns_distinct_both(self):

1603

obj1 = self.bus.get_object("busname1", "objectpath")

1604

self.assertIsInstance(obj1, Unique)

1605

obj2 = self.bus.get_object("busname2", "objectpath")

1606

self.assertIsInstance(obj2, Unique)

1607

self.assertIsNot(obj1, obj2)

1608

1609

def test_returns_same(self):

1610

obj1 = self.bus.get_object("busname", "objectpath")

1611

self.assertIsInstance(obj1, Unique)

1612

obj2 = self.bus.get_object("busname", "objectpath")

1613

self.assertIsInstance(obj2, Unique)

1614

self.assertIs(obj1, obj2)

1615

1616

def test_returns_same_old(self):

1617

obj1 = self.bus.get_object("busname1", "objectpath1")

1618

self.assertIsInstance(obj1, Unique)

1619

obj2 = self.bus.get_object("busname2", "objectpath2")

1620

self.assertIsInstance(obj2, Unique)

1621

obj1b = self.bus.get_object("busname1", "objectpath1")

1622

self.assertIsInstance(obj1b, Unique)

1623

self.assertIsNot(obj1, obj2)

1624

self.assertIsNot(obj2, obj1b)

1625

self.assertIs(obj1, obj1b)

1626

1627

1628

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1629

1630

def Stub_pydbus_func(self, func):

1631

class stub_pydbus:

1632

"""stub pydbus module"""

1633

class SystemBus:

1634

@staticmethod

1635

def get(busname, objectpath):

1636

DBusObject = collections.namedtuple(

1637

"DBusObject", ("methodname",))

1638

return {"interface":

1639

DBusObject(methodname=func)}

1640

return stub_pydbus

1641

1642

def call_method(self, bus, methodname, busname, objectpath,

1643

interface, *args):

1644

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

1645

return bus.call_method(methodname, busname, objectpath,

1646

interface, *args)

1647

1648

def test_call_method_returns(self):

1649

expected_method_return = Unique()

1650

method_args = (Unique(), Unique())

1651

def func(*args):

1652

self.assertEqual(len(method_args), len(args))

1653

for marg, arg in zip(method_args, args):

1654

self.assertIs(marg, arg)

1655

return expected_method_return

1656

stub_pydbus = self.Stub_pydbus_func(func)

1657

bus = pydbus_adapter.SystemBus(stub_pydbus)

1658

ret = self.call_method(bus, "methodname", "busname",

1659

"objectpath", "interface",

1660

*method_args)

1661

self.assertIs(ret, expected_method_return)

1662

1663

def test_call_method_handles_exception(self):

1664

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

1665

1666

def func():

1667

raise gi.repository.GLib.Error()

1668

1669

stub_pydbus = self.Stub_pydbus_func(func)

1670

bus = pydbus_adapter.SystemBus(stub_pydbus)

1671

1672

with self.assertRaises(dbus.Error) as e:

1673

self.call_method(bus, "methodname", "busname",

1674

"objectpath", "interface")

1675

1676

self.assertNotIsInstance(e, dbus.ConnectFailed)

1677

1678

def test_get_converts_to_correct_exception(self):

1679

bus = pydbus_adapter.SystemBus(

1680

self.fake_pydbus_raises_exception_on_connect)

1681

with self.assertRaises(dbus.ConnectFailed):

1682

self.call_method(bus, "methodname", "busname",

1683

"objectpath", "interface")

1684

1685

class fake_pydbus_raises_exception_on_connect:

1686

"""fake dbus-python module"""

1687

@classmethod

1688

def SystemBus(cls):

1689

def get(busname, objectpath):

1690

raise gi.repository.GLib.Error()

1691

Bus = collections.namedtuple("Bus", ["get"])

1692

return Bus(get=get)

1693

1694

def test_set_property_uses_setattr(self):

1695

class Object:

1696

pass

1697

obj = Object()

1698

class pydbus_spy:

1699

class SystemBus:

1700

@staticmethod

1701

def get(busname, objectpath):

1702

return {"interface": obj}

1703

bus = pydbus_adapter.SystemBus(pydbus_spy)

1704

value = Unique()

1705

bus.set_property("busname", "objectpath", "interface", "key",

1706

value)

1707

self.assertIs(value, obj.key)

1708

1709

def test_get_suppresses_xml_deprecation_warning(self):

1710

if sys.version_info.major >= 3:

1711

return

1712

class stub_pydbus_get:

1713

class SystemBus:

1714

@staticmethod

1715

def get(busname, objectpath):

1716

warnings.warn_explicit(

1717

"deprecated", DeprecationWarning,

1718

"xml.etree.ElementTree", 0)

1719

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1720

with warnings.catch_warnings(record=True) as w:

1721

warnings.simplefilter("always")

1722

bus.get("busname", "objectpath")

1723

self.assertEqual(0, len(w))

1724

1725

1726

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1727

class stub_pydbus:

1728

"""stub pydbus module"""

1729

class SystemBus:

1730

@staticmethod

1731

def get(busname, objectpath):

1732

return Unique()

1733

1734

def setUp(self):

1735

self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)

1736

1737

def test_returns_distinct_objectpaths(self):

1738

obj1 = self.bus.get("busname", "objectpath1")

1739

self.assertIsInstance(obj1, Unique)

1740

obj2 = self.bus.get("busname", "objectpath2")

1741

self.assertIsInstance(obj2, Unique)

1742

self.assertIsNot(obj1, obj2)

1743

1744

def test_returns_distinct_busnames(self):

1745

obj1 = self.bus.get("busname1", "objectpath")

1746

self.assertIsInstance(obj1, Unique)

1747

obj2 = self.bus.get("busname2", "objectpath")

1748

self.assertIsInstance(obj2, Unique)

1749

self.assertIsNot(obj1, obj2)

1750

1751

def test_returns_distinct_both(self):

1752

obj1 = self.bus.get("busname1", "objectpath")

1753

self.assertIsInstance(obj1, Unique)

1754

obj2 = self.bus.get("busname2", "objectpath")

1755

self.assertIsInstance(obj2, Unique)

1756

self.assertIsNot(obj1, obj2)

1757

1758

def test_returns_same(self):

1759

obj1 = self.bus.get("busname", "objectpath")

1760

self.assertIsInstance(obj1, Unique)

1761

obj2 = self.bus.get("busname", "objectpath")

1762

self.assertIsInstance(obj2, Unique)

1763

self.assertIs(obj1, obj2)

1764

1765

def test_returns_same_old(self):

1766

obj1 = self.bus.get("busname1", "objectpath1")

1767

self.assertIsInstance(obj1, Unique)

1768

obj2 = self.bus.get("busname2", "objectpath2")

1769

self.assertIsInstance(obj2, Unique)

1770

obj1b = self.bus.get("busname1", "objectpath1")

1771

self.assertIsInstance(obj1b, Unique)

1772

self.assertIsNot(obj1, obj2)

1773

self.assertIsNot(obj2, obj1b)

1774

self.assertIs(obj1, obj1b)

1775

1776

1777

class Test_commands_from_options(unittest.TestCase):

1778

1779

def setUp(self):

1780

self.parser = argparse.ArgumentParser()

1781

add_command_line_options(self.parser)

1782

1783

def test_is_enabled(self):

1784

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

1785

command.IsEnabled)

1786

1787

def assert_command_from_args(self, args, command_cls, length=1,

1788

clients=None, **cmd_attrs):

1789

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

1790

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

1791

options = self.parser.parse_args(args)

1792

check_option_syntax(self.parser, options)

1793

commands = commands_from_options(options)

1794

self.assertEqual(length, len(commands))

1795

for command in commands:

1796

if isinstance(command, command_cls):

1797

break

1798

else:

1799

self.assertIsInstance(command, command_cls)

1800

if clients is not None:

1801

self.assertEqual(clients, options.client)

1802

for key, value in cmd_attrs.items():

1803

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

1804

1805

def assert_commands_from_args(self, args, commands, clients=None):

1806

for cmd in commands:

1807

self.assert_command_from_args(args, cmd,

1808

length=len(commands),

1809

clients=clients)

1810

1811

def test_is_enabled_short(self):

1812

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

1813

command.IsEnabled)

1814

1815

def test_approve(self):

1816

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

1817

command.Approve)

1818

1819

def test_approve_short(self):

1820

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

1821

command.Approve)

1822

1823

def test_deny(self):

1824

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

1825

command.Deny)

1826

1827

def test_deny_short(self):

1828

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

1829

1830

def test_remove(self):

1831

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

1832

command.Remove)

1833

1834

def test_deny_before_remove(self):

1835

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

1836

"client"])

1837

check_option_syntax(self.parser, options)

1838

commands = commands_from_options(options)

1839

self.assertEqual(2, len(commands))

1840

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

1841

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

1842

1843

def test_deny_before_remove_reversed(self):

1844

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

1845

"--all"])

1846

check_option_syntax(self.parser, options)

1847

commands = commands_from_options(options)

1848

self.assertEqual(2, len(commands))

1849

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

1850

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

1851

1852

def test_remove_short(self):

1853

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

1854

command.Remove)

1855

1856

def test_dump_json(self):

1857

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

1858

command.DumpJSON)

1859

1860

def test_enable(self):

1861

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

1862

command.Enable)

1863

1864

def test_enable_short(self):

1865

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

1866

command.Enable)

1867

1868

def test_disable(self):

1869

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

1870

command.Disable)

1871

1872

def test_disable_short(self):

1873

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

1874

command.Disable)

1875

1876

def test_bump_timeout(self):

1877

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

1878

command.BumpTimeout)

1879

1880

def test_bump_timeout_short(self):

1881

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

1882

command.BumpTimeout)

1883

1884

def test_start_checker(self):

1885

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

1886

command.StartChecker)

1887

1888

def test_stop_checker(self):

1889

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

1890

command.StopChecker)

1891

1892

def test_approve_by_default(self):

1893

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

1894

"client"],

1895

command.ApproveByDefault)

1896

1897

def test_deny_by_default(self):

1898

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

1899

command.DenyByDefault)

1900

1901

def test_checker(self):

1902

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

1903

command.SetChecker,

1904

value_to_set=":")

1905

1906

def test_checker_empty(self):

1907

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

1908

command.SetChecker,

1909

value_to_set="")

1910

1911

def test_checker_short(self):

1912

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

1913

command.SetChecker,

1914

value_to_set=":")

1915

1916

def test_host(self):

1917

self.assert_command_from_args(

1918

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

1919

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

1920

1921

def test_host_short(self):

1922

self.assert_command_from_args(

1923

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

1924

value_to_set="client.example.org")

1925

1926

def test_secret_devnull(self):

1927

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

1928

"client"], command.SetSecret,

1929

value_to_set=b"")

1930

1931

def test_secret_tempfile(self):

1932

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

1933

value = b"secret\0xyzzy\nbar"

1934

f.write(value)

1935

f.seek(0)

1936

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

1937

"client"],

1938

command.SetSecret,

1939

value_to_set=value)

1940

1941

def test_secret_devnull_short(self):

1942

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

1943

"client"], command.SetSecret,

1944

value_to_set=b"")

1945

1946

def test_secret_tempfile_short(self):

1947

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

1948

value = b"secret\0xyzzy\nbar"

1949

f.write(value)

1950

f.seek(0)

1951

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

1952

command.SetSecret,

1953

value_to_set=value)

1954

1955

def test_timeout(self):

1956

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

1957

command.SetTimeout,

1958

value_to_set=300000)

1959

1960

def test_timeout_short(self):

1961

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

1962

command.SetTimeout,

1963

value_to_set=300000)

1964

1965

def test_extended_timeout(self):

1966

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

1967

"client"],

1968

command.SetExtendedTimeout,

1969

value_to_set=900000)

1970

1971

def test_interval(self):

1972

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

1973

"client"], command.SetInterval,

1974

value_to_set=120000)

1975

1976

def test_interval_short(self):

1977

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

1978

command.SetInterval,

1979

value_to_set=120000)

1980

1981

def test_approval_delay(self):

1982

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

1983

"client"],

1984

command.SetApprovalDelay,

1985

value_to_set=30000)

1986

1987

def test_approval_duration(self):

1988

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

1989

"client"],

1990

command.SetApprovalDuration,

1991

value_to_set=1000)

1992

1993

def test_print_table(self):

1994

self.assert_command_from_args([], command.PrintTable,

1995

verbose=False)

1996

1997

def test_print_table_verbose(self):

1998

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

1999

command.PrintTable,

2000

verbose=True)

2001

2002

def test_print_table_verbose_short(self):

2003

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

2004

verbose=True)

2005

2006

2007

def test_manual_page_example_1(self):

2008

self.assert_command_from_args("",

2009

command.PrintTable,

2010

clients=[],

2011

verbose=False)

2012

2013

def test_manual_page_example_2(self):

2014

self.assert_command_from_args(

2015

"--verbose foo1.example.org foo2.example.org".split(),

2016

command.PrintTable, clients=["foo1.example.org",

2017

"foo2.example.org"],

2018

verbose=True)

2019

2020

def test_manual_page_example_3(self):

2021

self.assert_command_from_args("--enable --all".split(),

2022

command.Enable,

2023

clients=[])

2024

2025

def test_manual_page_example_4(self):

2026

self.assert_commands_from_args(

2027

("--timeout=PT5M --interval=PT1M foo1.example.org"

2028

" foo2.example.org").split(),

2029

[command.SetTimeout, command.SetInterval],

2030

clients=["foo1.example.org", "foo2.example.org"])

2031

2032

def test_manual_page_example_5(self):

2033

self.assert_command_from_args("--approve --all".split(),

2034

command.Approve,

2035

clients=[])

2036

2037

2038

class TestCommand(unittest.TestCase):

2039

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

2040

2041

class FakeMandosBus(dbus.MandosBus):

2042

def __init__(self, testcase):

2043

self.client_properties = {

2044

"Name": "foo",

2045

"KeyID": ("92ed150794387c03ce684574b1139a65"

2046

"94a34f895daaaf09fd8ea90a27cddb12"),

2047

"Secret": b"secret",

2048

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

2049

"Enabled": True,

2050

"Timeout": 300000,

2051

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

2052

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

2053

"Interval": 120000,

2054

"Fingerprint": ("778827225BA7DE539C5A"

2055

"7CFA59CFF7CDBD9A5920"),

2056

"CheckerRunning": False,

2057

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

2058

"ApprovalPending": False,

2059

"ApprovedByDefault": True,

2060

"LastApprovalRequest": "",

2061

"ApprovalDelay": 0,

2062

"ApprovalDuration": 1000,

2063

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

2064

"ExtendedTimeout": 900000,

2065

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

2066

"LastCheckerStatus": 0,

2067

}

2068

self.other_client_properties = {

2069

"Name": "barbar",

2070

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2071

"6ab612cff5ad227247e46c2b020f441c"),

2072

"Secret": b"secretbar",

2073

"Host": "192.0.2.3",

2074

"Enabled": True,

2075

"Timeout": 300000,

2076

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

2077

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

2078

"Interval": 120000,

2079

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2080

"F547B3A107558FCA3A27"),

2081

"CheckerRunning": True,

2082

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

2083

"ApprovalPending": False,

2084

"ApprovedByDefault": False,

2085

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

2086

"ApprovalDelay": 30000,

2087

"ApprovalDuration": 93785000,

2088

"Checker": ":",

2089

"ExtendedTimeout": 900000,

2090

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

2091

"LastCheckerStatus": -2,

2092

}

2093

self.clients = collections.OrderedDict(

2094

[

2095

("client_objectpath", self.client_properties),

2096

("other_client_objectpath",

2097

self.other_client_properties),

2098

])

2099

self.one_client = {"client_objectpath":

2100

self.client_properties}

2101

self.testcase = testcase

2102

self.calls = []

2103

2104

def call_method(self, methodname, busname, objectpath,

2105

interface, *args):

2106

self.testcase.assertEqual("se.recompile.Mandos", busname)

2107

self.calls.append((methodname, busname, objectpath,

2108

interface, args))

2109

if interface == "org.freedesktop.DBus.Properties":

2110

if methodname == "Set":

2111

self.testcase.assertEqual(3, len(args))

2112

interface, key, value = args

2113

self.testcase.assertEqual(

2114

"se.recompile.Mandos.Client", interface)

2115

self.clients[objectpath][key] = value

2116

return

2117

elif interface == "se.recompile.Mandos":

2118

self.testcase.assertEqual("RemoveClient", methodname)

2119

self.testcase.assertEqual(1, len(args))

2120

clientpath = args[0]

2121

del self.clients[clientpath]

2122

return

2123

elif interface == "se.recompile.Mandos.Client":

2124

if methodname == "Approve":

2125

self.testcase.assertEqual(1, len(args))

2126

return

2127

raise ValueError()

2128

2129

def setUp(self):

2130

self.bus = self.FakeMandosBus(self)

2131

2132

2133

class TestBaseCommands(TestCommand):

2134

2135

def test_IsEnabled_exits_successfully(self):

2136

with self.assertRaises(SystemExit) as e:

2137

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

2138

if e.exception.code is not None:

2139

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

2140

else:

2141

self.assertIsNone(e.exception.code)

2142

2143

def test_IsEnabled_exits_with_failure(self):

2144

self.bus.client_properties["Enabled"] = False

2145

with self.assertRaises(SystemExit) as e:

2146

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

2147

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

2148

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

2149

else:

2150

self.assertIsNotNone(e.exception.code)

2151

2152

def test_Approve(self):

2153

busname = "se.recompile.Mandos"

2154

client_interface = "se.recompile.Mandos.Client"

2155

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

2156

for clientpath in self.bus.clients:

2157

self.assertIn(("Approve", busname, clientpath,

2158

client_interface, (True,)), self.bus.calls)

2159

2160

def test_Deny(self):

2161

busname = "se.recompile.Mandos"

2162

client_interface = "se.recompile.Mandos.Client"

2163

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

2164

for clientpath in self.bus.clients:

2165

self.assertIn(("Approve", busname, clientpath,

2166

client_interface, (False,)),

2167

self.bus.calls)

2168

2169

def test_Remove(self):

2170

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

2171

for clientpath in self.bus.clients:

2172

self.assertIn(("RemoveClient", dbus_busname,

2173

dbus_server_path, dbus_server_interface,

2174

(clientpath,)), self.bus.calls)

2175

2176

expected_json = {

2177

"foo": {

2178

"Name": "foo",

2179

"KeyID": ("92ed150794387c03ce684574b1139a65"

2180

"94a34f895daaaf09fd8ea90a27cddb12"),

2181

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

2182

"Enabled": True,

2183

"Timeout": 300000,

2184

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

2185

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

2186

"Interval": 120000,

2187

"Fingerprint": ("778827225BA7DE539C5A"

2188

"7CFA59CFF7CDBD9A5920"),

2189

"CheckerRunning": False,

2190

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

2191

"ApprovalPending": False,

2192

"ApprovedByDefault": True,

2193

"LastApprovalRequest": "",

2194

"ApprovalDelay": 0,

2195

"ApprovalDuration": 1000,

2196

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

2197

"ExtendedTimeout": 900000,

2198

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

2199

"LastCheckerStatus": 0,

2200

},

2201

"barbar": {

2202

"Name": "barbar",

2203

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2204

"6ab612cff5ad227247e46c2b020f441c"),

2205

"Host": "192.0.2.3",

2206

"Enabled": True,

2207

"Timeout": 300000,

2208

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

2209

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

2210

"Interval": 120000,

2211

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2212

"F547B3A107558FCA3A27"),

2213

"CheckerRunning": True,

2214

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

2215

"ApprovalPending": False,

2216

"ApprovedByDefault": False,

2217

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

2218

"ApprovalDelay": 30000,

2219

"ApprovalDuration": 93785000,

2220

"Checker": ":",

2221

"ExtendedTimeout": 900000,

2222

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

2223

"LastCheckerStatus": -2,

2224

},

2225

}

2226

2227

def test_DumpJSON_normal(self):

2228

with self.capture_stdout_to_buffer() as buffer:

2229

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

2230

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

2231

self.assertDictEqual(self.expected_json, json_data)

2232

2233

@staticmethod

2234

@contextlib.contextmanager

2235

def capture_stdout_to_buffer():

2236

capture_buffer = io.StringIO()

2237

old_stdout = sys.stdout

2238

sys.stdout = capture_buffer

2239

try:

2240

yield capture_buffer

2241

finally:

2242

sys.stdout = old_stdout

2243

2244

def test_DumpJSON_one_client(self):

2245

with self.capture_stdout_to_buffer() as buffer:

2246

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

2247

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

2248

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

2249

self.assertDictEqual(expected_json, json_data)

2250

2251

def test_PrintTable_normal(self):

2252

with self.capture_stdout_to_buffer() as buffer:

2253

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

2254

expected_output = "\n".join((

2255

"Name Enabled Timeout Last Successful Check",

2256

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

2257

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

2258

)) + "\n"

2259

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

2260

2261

def test_PrintTable_verbose(self):

2262

with self.capture_stdout_to_buffer() as buffer:

2263

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

2264

columns = (

2265

(

2266

"Name ",

2267

"foo ",

2268

"barbar ",

2269

),(

2270

"Enabled ",

2271

"Yes ",

2272

"Yes ",

2273

),(

2274

"Timeout ",

2275

"00:05:00 ",

2276

"00:05:00 ",

2277

),(

2278

"Last Successful Check ",

2279

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

2280

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

2281

),(

2282

"Created ",

2283

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

2284

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

2285

),(

2286

"Interval ",

2287

"00:02:00 ",

2288

"00:02:00 ",

2289

),(

2290

"Host ",

2291

"foo.example.org ",

2292

"192.0.2.3 ",

2293

),(

2294

("Key ID "

2295

" "),

2296

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2297

"ea90a27cddb12 "),

2298

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2299

"46c2b020f441c "),

2300

),(

2301

"Fingerprint ",

2302

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2303

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2304

),(

2305

"Check Is Running ",

2306

"No ",

2307

"Yes ",

2308

),(

2309

"Last Enabled ",

2310

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

2311

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

2312

),(

2313

"Approval Is Pending ",

2314

"No ",

2315

"No ",

2316

),(

2317

"Approved By Default ",

2318

"Yes ",

2319

"No ",

2320

),(

2321

"Last Approval Request ",

2322

" ",

2323

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

2324

),(

2325

"Approval Delay ",

2326

"00:00:00 ",

2327

"00:00:30 ",

2328

),(

2329

"Approval Duration ",

2330

"00:00:01 ",

2331

"1T02:03:05 ",

2332

),(

2333

"Checker ",

2334

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

2335

": ",

2336

),(

2337

"Extended Timeout ",

2338

"00:15:00 ",

2339

"00:15:00 ",

2340

),(

2341

"Expires ",

2342

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

2343

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

2344

),(

2345

"Last Checker Status",

2346

"0 ",

2347

"-2 ",

2348

)

2349

)

2350

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

2351

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

2352

for rows in columns)

2353

for line in range(num_lines))

2354

+ "\n")

2355

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

2356

2357

def test_PrintTable_one_client(self):

2358

with self.capture_stdout_to_buffer() as buffer:

2359

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

2360

expected_output = "\n".join((

2361

"Name Enabled Timeout Last Successful Check",

2362

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

2363

)) + "\n"

2364

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

2365

2366

2367

class TestPropertySetterCmd(TestCommand):

2368

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

2369

2370

def runTest(self):

2371

if not hasattr(self, "command"):

2372

return # Abstract TestCase class

2373

2374

if hasattr(self, "values_to_set"):

2375

cmd_args = [(value,) for value in self.values_to_set]

2376

values_to_get = getattr(self, "values_to_get",

2377

self.values_to_set)

2378

else:

2379

cmd_args = [() for x in range(len(self.values_to_get))]

2380

values_to_get = self.values_to_get

2381

for value_to_get, cmd_arg in zip(values_to_get, cmd_args):

2382

for clientpath in self.bus.clients:

2383

self.bus.clients[clientpath][self.propname] = (

2384

Unique())

2385

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

2386

for clientpath in self.bus.clients:

2387

value = (self.bus.clients[clientpath]

2388

[self.propname])

2389

self.assertNotIsInstance(value, Unique)

2390

self.assertEqual(value_to_get, value)

2391

2392

2393

class TestEnableCmd(TestPropertySetterCmd):

2394

command = command.Enable

2395

propname = "Enabled"

2396

values_to_get = [True]

2397

2398

2399

class TestDisableCmd(TestPropertySetterCmd):

2400

command = command.Disable

2401

propname = "Enabled"

2402

values_to_get = [False]

2403

2404

2405

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2406

command = command.BumpTimeout

2407

propname = "LastCheckedOK"

2408

values_to_get = [""]

2409

2410

2411

class TestStartCheckerCmd(TestPropertySetterCmd):

2412

command = command.StartChecker

2413

propname = "CheckerRunning"

2414

values_to_get = [True]

2415

2416

2417

class TestStopCheckerCmd(TestPropertySetterCmd):

2418

command = command.StopChecker

2419

propname = "CheckerRunning"

2420

values_to_get = [False]

2421

2422

2423

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2424

command = command.ApproveByDefault

2425

propname = "ApprovedByDefault"

2426

values_to_get = [True]

2427

2428

2429

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2430

command = command.DenyByDefault

2431

propname = "ApprovedByDefault"

2432

values_to_get = [False]

2433

2434

2435

class TestSetCheckerCmd(TestPropertySetterCmd):

2436

command = command.SetChecker

2437

propname = "Checker"

2438

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

2439

2440

2441

class TestSetHostCmd(TestPropertySetterCmd):

2442

command = command.SetHost

2443

propname = "Host"

2444

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

2445

2446

2447

class TestSetSecretCmd(TestPropertySetterCmd):

2448

command = command.SetSecret

2449

propname = "Secret"

2450

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

2451

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

2452

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

2453

2454

2455

class TestSetTimeoutCmd(TestPropertySetterCmd):

2456

command = command.SetTimeout

2457

propname = "Timeout"

2458

values_to_set = [datetime.timedelta(),

2459

datetime.timedelta(minutes=5),

2460

datetime.timedelta(seconds=1),

2461

datetime.timedelta(weeks=1),

2462

datetime.timedelta(weeks=52)]

2463

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

2464

2465

2466

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2467

command = command.SetExtendedTimeout

2468

propname = "ExtendedTimeout"

2469

values_to_set = [datetime.timedelta(),

2470

datetime.timedelta(minutes=5),

2471

datetime.timedelta(seconds=1),

2472

datetime.timedelta(weeks=1),

2473

datetime.timedelta(weeks=52)]

2474

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

2475

2476

2477

class TestSetIntervalCmd(TestPropertySetterCmd):

2478

command = command.SetInterval

2479

propname = "Interval"

2480

values_to_set = [datetime.timedelta(),

2481

datetime.timedelta(minutes=5),

2482

datetime.timedelta(seconds=1),

2483

datetime.timedelta(weeks=1),

2484

datetime.timedelta(weeks=52)]

2485

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

2486

2487

2488

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2489

command = command.SetApprovalDelay

2490

propname = "ApprovalDelay"

2491

values_to_set = [datetime.timedelta(),

2492

datetime.timedelta(minutes=5),

2493

datetime.timedelta(seconds=1),

2494

datetime.timedelta(weeks=1),

2495

datetime.timedelta(weeks=52)]

2496

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

2497

2498

2499

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2500

command = command.SetApprovalDuration

2501

propname = "ApprovalDuration"

2502

values_to_set = [datetime.timedelta(),

2503

datetime.timedelta(minutes=5),

2504

datetime.timedelta(seconds=1),

2505

datetime.timedelta(weeks=1),

2506

datetime.timedelta(weeks=52)]

2507

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

2508

2509

2510

2511

def should_only_run_tests():

2512

parser = argparse.ArgumentParser(add_help=False)

2513

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

2514

args, unknown_args = parser.parse_known_args()

2515

run_tests = args.check

2516

if run_tests:

2517

# Remove --check argument from sys.argv

2518

sys.argv[1:] = unknown_args

2519

return run_tests

2520

2521

# Add all tests from doctest strings

2522

def load_tests(loader, tests, none):

2523

import doctest

2524

tests.addTests(doctest.DocTestSuite())

2525

return tests

2526

2527

if __name__ == "__main__":

2528

try:

2529

if should_only_run_tests():

2530

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

2531

unittest.main()

2532

else:

2533

main()

2534

finally:

2535

logging.shutdown()