/mandos/trunk : revision 1103

1

#!/usr/bin/python

2

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

3

4

from __future__ import division

2

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

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This 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

40

import os

12

13

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

14

15

tablewords = {

16

'Name': u'Name',

17

'Enabled': u'Enabled',

18

'Timeout': u'Timeout',

19

'LastCheckedOK': u'Last Successful Check',

20

'Created': u'Created',

21

'Interval': u'Interval',

22

'Host': u'Host',

23

'Fingerprint': u'Fingerprint',

24

'CheckerRunning': u'Check Is Running',

25

'LastEnabled': u'Last Enabled',

26

'Checker': u'Checker',

27

}

28

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

29

domain = 'se.bsnet.fukt'

30

busname = domain + '.Mandos'

31

server_path = '/'

32

server_interface = domain + '.Mandos'

33

client_interface = domain + '.Mandos.Client'

34

version = "1.0.14"

35

36

def timedelta_to_milliseconds(td):

37

"Convert a datetime.timedelta object to milliseconds"

38

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

39

+ (td.seconds * 1000)

40

+ (td.microseconds // 1000))

41

42

def milliseconds_to_string(ms):

43

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

44

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

45

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

46

"hours": td.seconds // 3600,

47

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

48

"seconds": td.seconds % 60,

49

})

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

try:

50

import pydbus

51

import gi

52

dbus_python = None

53

except ImportError:

54

import dbus as dbus_python

55

pydbus = None

56

class gi(object):

57

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

58

class repository(object):

59

class GLib(object):

60

class Error(Exception):

61

pass

62

63

# Show warnings by default

64

if not sys.warnoptions:

65

import warnings

66

warnings.simplefilter("default")

67

68

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

69

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

70

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

71

72

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

73

74

if sys.version_info.major == 2:

75

str = unicode

76

import StringIO

77

io.StringIO = StringIO.StringIO

78

79

locale.setlocale(locale.LC_ALL, "")

80

81

version = "1.8.4"

82

83

84

def main():

85

parser = argparse.ArgumentParser()

86

add_command_line_options(parser)

87

88

options = parser.parse_args()

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

if pydbus is not None:

97

bus = pydbus_adapter.CachingBus(pydbus)

98

else:

99

bus = dbus_python_adapter.CachingBus(dbus_python)

100

101

try:

102

all_clients = bus.get_clients_and_properties()

103

except dbus.ConnectFailed as e:

104

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

105

sys.exit(1)

106

except dbus.Error as e:

107

log.critical(

108

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

109

sys.exit(1)

110

111

# Compile dict of (clientpath: properties) to process

112

if not clientnames:

113

clients = all_clients

114

else:

115

clients = {}

116

for name in clientnames:

117

for objpath, properties in all_clients.items():

118

if properties["Name"] == name:

119

clients[objpath] = properties

120

break

121

else:

122

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

123

sys.exit(1)

124

125

commands = commands_from_options(options)

126

127

for command in commands:

128

command.run(clients, bus)

129

130

131

def add_command_line_options(parser):

132

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

133

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

134

help="show version number and exit")

135

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

136

help="Select all clients")

137

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

138

help="Print all fields")

139

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

140

action="append_const", default=[],

141

const=command.DumpJSON(),

142

help="Dump client data in JSON format")

143

enable_disable = parser.add_mutually_exclusive_group()

144

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

145

action="append_const", default=[],

146

const=command.Enable(),

147

help="Enable client")

148

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

149

action="append_const", default=[],

150

const=command.Disable(),

151

help="disable client")

152

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

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

155

help="Bump timeout for client")

156

start_stop_checker = parser.add_mutually_exclusive_group()

157

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

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

161

help="Start checker for client")

162

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

163

action="append_const", default=[],

164

const=command.StopChecker(),

165

help="Stop checker for client")

166

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

167

action="append_const", default=[],

168

const=command.IsEnabled(),

169

help="Check if client is enabled")

170

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

171

action="append_const", default=[],

172

const=command.Remove(),

173

help="Remove client")

174

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

175

action="append", default=[],

176

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

177

help="Set checker command for client")

178

parser.add_argument(

179

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

180

default=[], metavar="TIME",

181

type=command.SetTimeout.argparse(string_to_delta),

182

help="Set timeout for client")

183

parser.add_argument(

184

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

185

default=[], metavar="TIME",

186

type=command.SetExtendedTimeout.argparse(string_to_delta),

187

help="Set extended timeout for client")

188

parser.add_argument(

189

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

190

default=[], metavar="TIME",

191

type=command.SetInterval.argparse(string_to_delta),

192

help="Set checker interval for client")

193

approve_deny_default = parser.add_mutually_exclusive_group()

194

approve_deny_default.add_argument(

195

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

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

198

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

199

approve_deny_default.add_argument(

200

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

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

203

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

204

parser.add_argument(

205

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

206

default=[], metavar="TIME",

207

type=command.SetApprovalDelay.argparse(string_to_delta),

208

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

209

parser.add_argument(

210

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

211

default=[], metavar="TIME",

212

type=command.SetApprovalDuration.argparse(string_to_delta),

213

help="Set duration of one client approval")

214

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

215

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

216

type=command.SetHost,

217

help="Set host for client")

218

parser.add_argument(

219

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

220

default=[], metavar="FILENAME",

221

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

222

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

223

approve_deny = parser.add_mutually_exclusive_group()

224

approve_deny.add_argument(

225

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

226

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

227

help="Approve any current client request")

228

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

229

action="append_const", default=[],

230

const=command.Deny(),

231

help="Deny any current client request")

232

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

233

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

234

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

235

help="Run self-test")

236

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

50

237

51

238

52

239

def string_to_delta(interval):

53

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

54

55

>>> string_to_delta('7d')

240

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

241

242

try:

243

return rfc3339_duration_to_delta(interval)

244

except ValueError as e:

245

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

246

' '.join(e.args))

247

return parse_pre_1_6_1_interval(interval)

248

249

250

def rfc3339_duration_to_delta(duration):

251

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

252

253

>>> rfc3339_duration_to_delta("P7D")

56

254

datetime.timedelta(7)

57

>>> string_to_delta('60s')

255

>>> rfc3339_duration_to_delta("PT60S")

58

256

datetime.timedelta(0, 60)

59

>>> string_to_delta('60m')

257

>>> rfc3339_duration_to_delta("PT60M")

60

258

datetime.timedelta(0, 3600)

61

>>> string_to_delta('24h')

259

>>> rfc3339_duration_to_delta("P60M")

260

datetime.timedelta(1680)

261

>>> rfc3339_duration_to_delta("PT24H")

62

262

datetime.timedelta(1)

63

>>> string_to_delta(u'1w')

263

>>> rfc3339_duration_to_delta("P1W")

64

264

datetime.timedelta(7)

65

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

265

>>> rfc3339_duration_to_delta("PT5M30S")

66

266

datetime.timedelta(0, 330)

267

>>> rfc3339_duration_to_delta("P1DT3M20S")

268

datetime.timedelta(1, 200)

269

>>> # Can not be empty:

270

>>> rfc3339_duration_to_delta("")

271

Traceback (most recent call last):

272

...

273

ValueError: Invalid RFC 3339 duration: ""

274

>>> # Must start with "P":

275

>>> rfc3339_duration_to_delta("1D")

276

Traceback (most recent call last):

277

...

278

ValueError: Invalid RFC 3339 duration: "1D"

279

>>> # Must use correct order

280

>>> rfc3339_duration_to_delta("PT1S2M")

281

Traceback (most recent call last):

282

...

283

ValueError: Invalid RFC 3339 duration: "PT1S2M"

284

>>> # Time needs time marker

285

>>> rfc3339_duration_to_delta("P1H2S")

286

Traceback (most recent call last):

287

...

288

ValueError: Invalid RFC 3339 duration: "P1H2S"

289

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

290

>>> rfc3339_duration_to_delta("P1D2W")

291

Traceback (most recent call last):

292

...

293

ValueError: Invalid RFC 3339 duration: "P1D2W"

294

>>> rfc3339_duration_to_delta("P2W2H")

295

Traceback (most recent call last):

296

...

297

ValueError: Invalid RFC 3339 duration: "P2W2H"

67

298

"""

68

timevalue = datetime.timedelta(0)

69

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

70

71

for s in regexp.findall(interval):

72

try:

73

suffix = unicode(s[-1])

74

value = int(s[:-1])

75

if suffix == u"d":

76

delta = datetime.timedelta(value)

77

elif suffix == u"s":

78

delta = datetime.timedelta(0, value)

79

elif suffix == u"m":

80

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

81

elif suffix == u"h":

82

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

83

elif suffix == u"w":

84

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

85

else:

86

raise ValueError

87

except (ValueError, IndexError):

88

raise ValueError

89

timevalue += delta

90

return timevalue

91

92

def print_clients(clients, keywords):

93

def valuetostring(value, keyword):

94

if type(value) is dbus.Boolean:

95

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

96

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

97

return milliseconds_to_string(value)

98

return unicode(value)

99

100

# Create format string to print table rows

101

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

102

max(len(tablewords[key]),

103

max(len(valuetostring(client[key],

104

key))

105

for client in

106

clients))

107

for key in keywords)

108

# Print header line

109

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

110

for client in clients:

111

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

112

for key in keywords)

113

def has_actions(options):

114

return any((options.enable,

115

options.disable,

116

options.bump_timeout,

117

options.start_checker,

118

options.stop_checker,

119

options.is_enabled,

120

options.remove,

121

options.checker is not None,

122

options.timeout is not None,

123

options.interval is not None,

124

options.host is not None,

125

options.secret is not None,

126

options.approve,

127

options.deny))

128

129

def main():

130

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

131

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

132

help="Select all clients")

133

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

134

help="Print all fields")

135

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

136

help="Enable client")

137

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

138

help="disable client")

139

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

140

help="Bump timeout for client")

141

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

142

help="Start checker for client")

143

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

144

help="Stop checker for client")

145

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

146

help="Check if client is enabled")

147

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

148

help="Remove client")

149

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

150

help="Set checker command for client")

151

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

152

help="Set timeout for client")

153

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

154

help="Set checker interval for client")

155

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

156

help="Set host for client")

157

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

158

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

159

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

160

help="Approve any current client request")

161

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

162

help="Deny any current client request")

163

options, client_names = parser.parse_args()

164

165

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

166

parser.error('Options requires clients names or --all.')

167

if options.verbose and has_actions(options):

168

parser.error('Verbose option can only be used alone or with --all.')

169

if options.all and not has_actions(options):

170

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

171

172

try:

173

bus = dbus.SystemBus()

174

mandos_dbus_objc = bus.get_object(busname, server_path)

175

except dbus.exceptions.DBusException:

176

print >> sys.stderr, "Could not connect to Mandos server"

177

sys.exit(1)

178

179

mandos_serv = dbus.Interface(mandos_dbus_objc,

180

dbus_interface = server_interface)

181

182

#block stderr since dbus library prints to stderr

183

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

184

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

185

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

186

os.close(null)

187

try:

188

try:

189

mandos_clients = mandos_serv.GetAllClientsWithProperties()

190

finally:

191

#restore stderr

192

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

193

os.close(stderrcopy)

194

except dbus.exceptions.DBusException, e:

195

print >> sys.stderr, "Access denied: Accessing mandos server through dbus."

196

sys.exit(1)

197

198

# Compile list of clients to process

199

clients=[]

200

201

if options.all or not client_names:

202

clients = (bus.get_object(busname, path) for path in mandos_clients.iterkeys())

299

300

# Parsing an RFC 3339 duration with regular expressions is not

301

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

302

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

303

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

304

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

305

# avoid excessive use of external libraries.

306

307

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

308

Token = collections.namedtuple("Token", (

309

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

310

# a "group" containing digits

311

"value", # datetime.timedelta or None

312

"followers")) # Tokens valid after this token

313

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

314

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

315

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

316

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

317

datetime.timedelta(seconds=1),

318

frozenset((token_end, )))

319

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

320

datetime.timedelta(minutes=1),

321

frozenset((token_second, token_end)))

322

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

323

datetime.timedelta(hours=1),

324

frozenset((token_minute, token_end)))

325

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

326

None,

327

frozenset((token_hour, token_minute,

328

token_second)))

329

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

330

datetime.timedelta(days=1),

331

frozenset((token_time, token_end)))

332

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

333

datetime.timedelta(weeks=4),

334

frozenset((token_day, token_end)))

335

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

336

datetime.timedelta(weeks=52),

337

frozenset((token_month, token_end)))

338

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

339

datetime.timedelta(weeks=1),

340

frozenset((token_end, )))

341

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

342

frozenset((token_year, token_month,

343

token_day, token_time,

344

token_week)))

345

# Define starting values:

346

# Value so far

347

value = datetime.timedelta()

348

found_token = None

349

# Following valid tokens

350

followers = frozenset((token_duration, ))

351

# String left to parse

352

s = duration

353

# Loop until end token is found

354

while found_token is not token_end:

355

# Search for any currently valid tokens

356

for token in followers:

357

match = token.regexp.match(s)

358

if match is not None:

359

# Token found

360

if token.value is not None:

361

# Value found, parse digits

362

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

363

# Add to value so far

364

value += factor * token.value

365

# Strip token from string

366

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

367

# Go to found token

368

found_token = token

369

# Set valid next tokens

370

followers = found_token.followers

371

break

203

372

else:

204

for name in client_names:

205

for path, client in mandos_clients.iteritems():

206

if client['Name'] == name:

207

client_objc = bus.get_object(busname, path)

208

clients.append(client_objc)

209

break

373

# No currently valid tokens were found

374

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

375

.format(duration))

376

# End token found

377

return value

378

379

380

def parse_pre_1_6_1_interval(interval):

381

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

382

and return a datetime.timedelta

383

384

>>> parse_pre_1_6_1_interval('7d')

385

datetime.timedelta(7)

386

>>> parse_pre_1_6_1_interval('60s')

387

datetime.timedelta(0, 60)

388

>>> parse_pre_1_6_1_interval('60m')

389

datetime.timedelta(0, 3600)

390

>>> parse_pre_1_6_1_interval('24h')

391

datetime.timedelta(1)

392

>>> parse_pre_1_6_1_interval('1w')

393

datetime.timedelta(7)

394

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

395

datetime.timedelta(0, 330)

396

>>> parse_pre_1_6_1_interval('')

397

datetime.timedelta(0)

398

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

399

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

400

datetime.timedelta(2, 480, 18000)

401

402

"""

403

404

value = datetime.timedelta(0)

405

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

406

407

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

408

if suffix == "d":

409

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

410

elif suffix == "s":

411

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

412

elif suffix == "m":

413

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

414

elif suffix == "h":

415

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

416

elif suffix == "w":

417

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

418

elif suffix == "":

419

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

420

return value

421

422

423

def check_option_syntax(parser, options):

424

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

425

argparse"""

426

427

def has_commands(options, commands=None):

428

if commands is None:

429

commands = (command.Enable,

430

command.Disable,

431

command.BumpTimeout,

432

command.StartChecker,

433

command.StopChecker,

434

command.IsEnabled,

435

command.Remove,

436

command.SetChecker,

437

command.SetTimeout,

438

command.SetExtendedTimeout,

439

command.SetInterval,

440

command.ApproveByDefault,

441

command.DenyByDefault,

442

command.SetApprovalDelay,

443

command.SetApprovalDuration,

444

command.SetHost,

445

command.SetSecret,

446

command.Approve,

447

command.Deny)

448

return any(isinstance(cmd, commands)

449

for cmd in options.commands)

450

451

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

452

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

453

if options.verbose and has_commands(options):

454

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

455

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

456

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

457

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

458

if options.all and not has_commands(options):

459

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

460

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

461

and len(options.client) > 1):

462

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

463

if (len(options.commands) > 1

464

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

465

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

466

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

467

468

469

class dbus(object):

470

471

class SystemBus(object):

472

473

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

474

def get_managed_objects(self, busname, objectpath):

475

return self.call_method("GetManagedObjects", busname,

476

objectpath,

477

self.object_manager_iface)

478

479

properties_iface = "org.freedesktop.DBus.Properties"

480

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

481

value):

482

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

483

self.properties_iface, interface, key,

484

value)

485

486

487

class MandosBus(SystemBus):

488

busname_domain = "se.recompile"

489

busname = busname_domain + ".Mandos"

490

server_path = "/"

491

server_interface = busname_domain + ".Mandos"

492

client_interface = busname_domain + ".Mandos.Client"

493

del busname_domain

494

495

def get_clients_and_properties(self):

496

managed_objects = self.get_managed_objects(

497

self.busname, self.server_path)

498

return {objpath: properties[self.client_interface]

499

for objpath, properties in managed_objects.items()

500

if self.client_interface in properties}

501

502

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

503

return self.set_property(self.busname, objectpath,

504

self.client_interface, key,

505

value)

506

507

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

508

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

509

self.client_interface, *args)

510

511

def call_server_method(self, method, *args):

512

return self.call_method(method, self.busname,

513

self.server_path,

514

self.server_interface, *args)

515

516

class Error(Exception):

517

pass

518

519

class ConnectFailed(Error):

520

pass

521

522

523

class dbus_python_adapter(object):

524

525

class SystemBus(dbus.MandosBus):

526

"""Use dbus-python"""

527

528

def __init__(self, module=dbus_python):

529

self.dbus_python = module

530

self.bus = self.dbus_python.SystemBus()

531

532

@contextlib.contextmanager

533

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

534

try:

535

yield

536

except self.dbus_python.exceptions.DBusException as e:

537

# This does what "raise from" would do

538

exc = exception_class(*e.args)

539

exc.__cause__ = e

540

raise exc

541

542

def call_method(self, methodname, busname, objectpath,

543

interface, *args):

544

proxy_object = self.get_object(busname, objectpath)

545

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

546

interface, methodname,

547

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

548

method = getattr(proxy_object, methodname)

549

with self.convert_exception():

550

with dbus_python_adapter.SilenceLogger(

551

"dbus.proxies"):

552

value = method(*args, dbus_interface=interface)

553

return self.type_filter(value)

554

555

def get_object(self, busname, objectpath):

556

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

557

busname, objectpath)

558

with self.convert_exception(dbus.ConnectFailed):

559

return self.bus.get_object(busname, objectpath)

560

561

def type_filter(self, value):

562

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

563

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

564

return bool(value)

565

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

566

return str(value)

567

# Also recurse into dictionaries

568

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

569

return {self.type_filter(key):

570

self.type_filter(subval)

571

for key, subval in value.items()}

572

return value

573

574

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

575

if key == "Secret":

576

if not isinstance(value, bytes):

577

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

578

value = self.dbus_python.ByteArray(value)

579

return self.set_property(self.busname, objectpath,

580

self.client_interface, key,

581

value)

582

583

class SilenceLogger(object):

584

"Simple context manager to silence a particular logger"

585

def __init__(self, loggername):

586

self.logger = logging.getLogger(loggername)

587

588

def __enter__(self):

589

self.logger.addFilter(self.nullfilter)

590

591

class NullFilter(logging.Filter):

592

def filter(self, record):

593

return False

594

595

nullfilter = NullFilter()

596

597

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

598

self.logger.removeFilter(self.nullfilter)

599

600

601

class CachingBus(SystemBus):

602

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

603

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

604

self.object_cache = {}

605

super(dbus_python_adapter.CachingBus,

606

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

607

def get_object(self, busname, objectpath):

608

try:

609

return self.object_cache[(busname, objectpath)]

610

except KeyError:

611

new_object = super(

612

dbus_python_adapter.CachingBus,

613

self).get_object(busname, objectpath)

614

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

615

return new_object

616

617

618

class pydbus_adapter(object):

619

class SystemBus(dbus.MandosBus):

620

def __init__(self, module=pydbus):

621

self.pydbus = module

622

self.bus = self.pydbus.SystemBus()

623

624

@contextlib.contextmanager

625

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

626

try:

627

yield

628

except gi.repository.GLib.Error as e:

629

# This does what "raise from" would do

630

exc = exception_class(*e.args)

631

exc.__cause__ = e

632

raise exc

633

634

def call_method(self, methodname, busname, objectpath,

635

interface, *args):

636

proxy_object = self.get(busname, objectpath)

637

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

638

interface, methodname,

639

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

640

method = getattr(proxy_object[interface], methodname)

641

with self.convert_exception():

642

return method(*args)

643

644

def get(self, busname, objectpath):

645

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

646

busname, objectpath)

647

with self.convert_exception(dbus.ConnectFailed):

648

if sys.version_info.major <= 2:

649

with warnings.catch_warnings():

650

warnings.filterwarnings(

651

"ignore", "", DeprecationWarning,

652

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

653

return self.bus.get(busname, objectpath)

210

654

else:

211

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

212

sys.exit(1)

213

214

if not has_actions(options) and clients:

215

if options.verbose:

216

keywords = ('Name', 'Enabled', 'Timeout', 'LastCheckedOK',

217

'Created', 'Interval', 'Host', 'Fingerprint',

218

'CheckerRunning', 'LastEnabled', 'Checker')

219

else:

220

keywords = defaultkeywords

221

222

print_clients(mandos_clients.values(), keywords)

223

else:

224

# Process each client in the list by all selected options

225

for client in clients:

226

if options.remove:

227

mandos_serv.RemoveClient(client.__dbus_object_path__)

228

if options.enable:

229

client.Enable(dbus_interface=client_interface)

230

if options.disable:

231

client.Disable(dbus_interface=client_interface)

232

if options.bump_timeout:

233

client.CheckedOK(dbus_interface=client_interface)

234

if options.start_checker:

235

client.StartChecker(dbus_interface=client_interface)

236

if options.stop_checker:

237

client.StopChecker(dbus_interface=client_interface)

238

if options.is_enabled:

239

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

240

u"Enabled",

241

dbus_interface=dbus.PROPERTIES_IFACE)

242

else 1)

243

if options.checker:

244

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

245

dbus_interface=dbus.PROPERTIES_IFACE)

246

if options.host:

247

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

248

dbus_interface=dbus.PROPERTIES_IFACE)

249

if options.interval:

250

client.Set(client_interface, u"Interval",

251

timedelta_to_milliseconds

252

(string_to_delta(options.interval)),

253

dbus_interface=dbus.PROPERTIES_IFACE)

254

if options.timeout:

255

client.Set(client_interface, u"Timeout",

256

timedelta_to_milliseconds(string_to_delta

257

(options.timeout)),

258

dbus_interface=dbus.PROPERTIES_IFACE)

259

if options.secret:

260

client.Set(client_interface, u"Secret",

261

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

262

dbus_interface=dbus.PROPERTIES_IFACE)

263

if options.approve:

264

client.Approve(dbus.Boolean(True), dbus_interface=client_interface)

265

if options.deny:

266

client.Approve(dbus.Boolean(False), dbus_interface=client_interface)

267

268

if __name__ == '__main__':

269

main()

655

return self.bus.get(busname, objectpath)

656

657

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

658

value):

659

proxy_object = self.get(busname, objectpath)

660

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

661

objectpath, self.properties_iface, interface,

662

key, value)

663

setattr(proxy_object[interface], key, value)

664

665

class CachingBus(SystemBus):

666

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

667

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

668

self.object_cache = {}

669

super(pydbus_adapter.CachingBus,

670

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

671

def get(self, busname, objectpath):

672

try:

673

return self.object_cache[(busname, objectpath)]

674

except KeyError:

675

new_object = (super(pydbus_adapter.CachingBus, self)

676

.get(busname, objectpath))

677

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

678

return new_object

679

680

681

def commands_from_options(options):

682

683

commands = list(options.commands)

684

685

def find_cmd(cmd, commands):

686

i = 0

687

for i, c in enumerate(commands):

688

if isinstance(c, cmd):

689

return i

690

return i+1

691

692

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

693

# to occur ahead of command.Remove.

694

index_of_remove = find_cmd(command.Remove, commands)

695

before_remove = commands[:index_of_remove]

696

after_remove = commands[index_of_remove:]

697

cleaned_after = []

698

for cmd in after_remove:

699

if isinstance(cmd, command.Deny):

700

before_remove.append(cmd)

701

else:

702

cleaned_after.append(cmd)

703

if cleaned_after != after_remove:

704

commands = before_remove + cleaned_after

705

706

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

707

# optionally verbosely

708

if not commands:

709

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

710

711

return commands

712

713

714

class command(object):

715

"""A namespace for command classes"""

716

717

class Base(object):

718

"""Abstract base class for commands"""

719

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

720

"""Normal commands should implement run_on_one_client(),

721

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

722

override this run() method instead.

723

"""

724

self.bus = bus

725

for client, properties in clients.items():

726

self.run_on_one_client(client, properties)

727

728

729

class IsEnabled(Base):

730

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

731

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

732

if properties["Enabled"]:

733

sys.exit(0)

734

sys.exit(1)

735

736

737

class Approve(Base):

738

def run_on_one_client(self, client, properties):

739

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

740

741

742

class Deny(Base):

743

def run_on_one_client(self, client, properties):

744

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

745

746

747

class Remove(Base):

748

def run(self, clients, bus):

749

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

750

bus.call_server_method("RemoveClient", clientpath)

751

752

753

class Output(Base):

754

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

755

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

756

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

757

"Fingerprint", "CheckerRunning",

758

"LastEnabled", "ApprovalPending",

759

"ApprovedByDefault", "LastApprovalRequest",

760

"ApprovalDelay", "ApprovalDuration",

761

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

762

"LastCheckerStatus")

763

764

765

class DumpJSON(Output):

766

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

767

data = {properties["Name"]:

768

{key: properties[key]

769

for key in self.all_keywords}

770

for properties in clients.values()}

771

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

772

773

774

class PrintTable(Output):

775

def __init__(self, verbose=False):

776

self.verbose = verbose

777

778

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

779

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

780

"LastCheckedOK")

781

keywords = default_keywords

782

if self.verbose:

783

keywords = self.all_keywords

784

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

785

786

class TableOfClients(object):

787

tableheaders = {

788

"Name": "Name",

789

"Enabled": "Enabled",

790

"Timeout": "Timeout",

791

"LastCheckedOK": "Last Successful Check",

792

"LastApprovalRequest": "Last Approval Request",

793

"Created": "Created",

794

"Interval": "Interval",

795

"Host": "Host",

796

"Fingerprint": "Fingerprint",

797

"KeyID": "Key ID",

798

"CheckerRunning": "Check Is Running",

799

"LastEnabled": "Last Enabled",

800

"ApprovalPending": "Approval Is Pending",

801

"ApprovedByDefault": "Approved By Default",

802

"ApprovalDelay": "Approval Delay",

803

"ApprovalDuration": "Approval Duration",

804

"Checker": "Checker",

805

"ExtendedTimeout": "Extended Timeout",

806

"Expires": "Expires",

807

"LastCheckerStatus": "Last Checker Status",

808

}

809

810

def __init__(self, clients, keywords):

811

self.clients = clients

812

self.keywords = keywords

813

814

def __str__(self):

815

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

816

817

if sys.version_info.major == 2:

818

__unicode__ = __str__

819

def __str__(self):

820

return str(self).encode(

821

locale.getpreferredencoding())

822

823

def rows(self):

824

format_string = self.row_formatting_string()

825

rows = [self.header_line(format_string)]

826

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

827

for client in self.clients)

828

return rows

829

830

def row_formatting_string(self):

831

"Format string used to format table rows"

832

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

833

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

834

*(len(self.string_from_client(client,

835

key))

836

for client in self.clients)),

837

key=key)

838

for key in self.keywords)

839

840

def string_from_client(self, client, key):

841

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

842

843

@classmethod

844

def valuetostring(cls, value, keyword):

845

if isinstance(value, bool):

846

return "Yes" if value else "No"

847

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

848

"ApprovalDuration", "ExtendedTimeout"):

849

return cls.milliseconds_to_string(value)

850

return str(value)

851

852

def header_line(self, format_string):

853

return format_string.format(**self.tableheaders)

854

855

def client_line(self, client, format_string):

856

return format_string.format(

857

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

858

for key in self.keywords})

859

860

@staticmethod

861

def milliseconds_to_string(ms):

862

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

863

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

864

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

865

if td.days else "",

866

hours=td.seconds // 3600,

867

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

868

seconds=td.seconds % 60))

869

870

871

class PropertySetter(Base):

872

"Abstract class for Actions for setting one client property"

873

874

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

875

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

876

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

877

self.value_to_set)

878

879

@property

880

def propname(self):

881

raise NotImplementedError()

882

883

884

class Enable(PropertySetter):

885

propname = "Enabled"

886

value_to_set = True

887

888

889

class Disable(PropertySetter):

890

propname = "Enabled"

891

value_to_set = False

892

893

894

class BumpTimeout(PropertySetter):

895

propname = "LastCheckedOK"

896

value_to_set = ""

897

898

899

class StartChecker(PropertySetter):

900

propname = "CheckerRunning"

901

value_to_set = True

902

903

904

class StopChecker(PropertySetter):

905

propname = "CheckerRunning"

906

value_to_set = False

907

908

909

class ApproveByDefault(PropertySetter):

910

propname = "ApprovedByDefault"

911

value_to_set = True

912

913

914

class DenyByDefault(PropertySetter):

915

propname = "ApprovedByDefault"

916

value_to_set = False

917

918

919

class PropertySetterValue(PropertySetter):

920

"""Abstract class for PropertySetter recieving a value as

921

constructor argument instead of a class attribute."""

922

def __init__(self, value):

923

self.value_to_set = value

924

925

@classmethod

926

def argparse(cls, argtype):

927

def cmdtype(arg):

928

return cls(argtype(arg))

929

return cmdtype

930

931

class SetChecker(PropertySetterValue):

932

propname = "Checker"

933

934

935

class SetHost(PropertySetterValue):

936

propname = "Host"

937

938

939

class SetSecret(PropertySetterValue):

940

propname = "Secret"

941

942

@property

943

def value_to_set(self):

944

return self._vts

945

946

@value_to_set.setter

947

def value_to_set(self, value):

948

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

949

self._vts = value.read()

950

value.close()

951

952

953

class PropertySetterValueMilliseconds(PropertySetterValue):

954

"""Abstract class for PropertySetterValue taking a value

955

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

956

milliseconds."""

957

958

@property

959

def value_to_set(self):

960

return self._vts

961

962

@value_to_set.setter

963

def value_to_set(self, value):

964

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

965

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

966

967

968

class SetTimeout(PropertySetterValueMilliseconds):

969

propname = "Timeout"

970

971

972

class SetExtendedTimeout(PropertySetterValueMilliseconds):

973

propname = "ExtendedTimeout"

974

975

976

class SetInterval(PropertySetterValueMilliseconds):

977

propname = "Interval"

978

979

980

class SetApprovalDelay(PropertySetterValueMilliseconds):

981

propname = "ApprovalDelay"

982

983

984

class SetApprovalDuration(PropertySetterValueMilliseconds):

985

propname = "ApprovalDuration"

986

987

988

989

class TestCaseWithAssertLogs(unittest.TestCase):

990

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

991

992

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

993

@contextlib.contextmanager

994

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

995

capturing_handler = self.CapturingLevelHandler(level)

996

old_level = logger.level

997

old_propagate = logger.propagate

998

logger.addHandler(capturing_handler)

999

logger.setLevel(level)

1000

logger.propagate = False

1001

try:

1002

yield capturing_handler.watcher

1003

finally:

1004

logger.propagate = old_propagate

1005

logger.removeHandler(capturing_handler)

1006

logger.setLevel(old_level)

1007

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

1008

0)

1009

1010

class CapturingLevelHandler(logging.Handler):

1011

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

1012

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

1013

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

1014

def emit(self, record):

1015

self.watcher.records.append(record)

1016

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

1017

1018

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1019

("records",

1020

"output"))

1021

1022

1023

class Unique(object):

1024

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

1025

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

1026

1027

1028

class Test_string_to_delta(TestCaseWithAssertLogs):

1029

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

1030

# rfc3339_duration_to_delta() already has more comprehensive

1031

# tests, which are run by doctest.

1032

1033

def test_rfc3339_zero_seconds(self):

1034

self.assertEqual(datetime.timedelta(),

1035

string_to_delta("PT0S"))

1036

1037

def test_rfc3339_zero_days(self):

1038

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

1039

1040

def test_rfc3339_one_second(self):

1041

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

1042

string_to_delta("PT1S"))

1043

1044

def test_rfc3339_two_hours(self):

1045

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

1046

string_to_delta("PT2H"))

1047

1048

def test_falls_back_to_pre_1_6_1_with_warning(self):

1049

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

1050

value = string_to_delta("2h")

1051

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

1052

1053

1054

class Test_check_option_syntax(unittest.TestCase):

1055

def setUp(self):

1056

self.parser = argparse.ArgumentParser()

1057

add_command_line_options(self.parser)

1058

1059

def test_actions_requires_client_or_all(self):

1060

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

1061

args = self.actionargs(action, value)

1062

with self.assertParseError():

1063

self.parse_args(args)

1064

1065

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

1066

# check_option_syntax()

1067

actions = {

1068

"--enable": None,

1069

"--disable": None,

1070

"--bump-timeout": None,

1071

"--start-checker": None,

1072

"--stop-checker": None,

1073

"--is-enabled": None,

1074

"--remove": None,

1075

"--checker": "x",

1076

"--timeout": "PT0S",

1077

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

1078

"--interval": "PT0S",

1079

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

1080

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

1081

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

1082

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

1083

"--host": "hostname",

1084

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

1085

"--approve": None,

1086

"--deny": None,

1087

}

1088

1089

@staticmethod

1090

def actionargs(action, value, *args):

1091

if value is not None:

1092

return [action, value] + list(args)

1093

else:

1094

return [action] + list(args)

1095

1096

@contextlib.contextmanager

1097

def assertParseError(self):

1098

with self.assertRaises(SystemExit) as e:

1099

with self.redirect_stderr_to_devnull():

1100

yield

1101

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

1102

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

1103

# /argparse.html#exiting-methods

1104

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

1105

1106

def parse_args(self, args):

1107

options = self.parser.parse_args(args)

1108

check_option_syntax(self.parser, options)

1109

1110

@staticmethod

1111

@contextlib.contextmanager

1112

def redirect_stderr_to_devnull():

1113

old_stderr = sys.stderr

1114

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

1115

sys.stderr = null

1116

try:

1117

yield

1118

finally:

1119

sys.stderr = old_stderr

1120

1121

def check_option_syntax(self, options):

1122

check_option_syntax(self.parser, options)

1123

1124

def test_actions_all_conflicts_with_verbose(self):

1125

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

1126

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

1127

"--verbose")

1128

with self.assertParseError():

1129

self.parse_args(args)

1130

1131

def test_actions_with_client_conflicts_with_verbose(self):

1132

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

1133

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

1134

"client")

1135

with self.assertParseError():

1136

self.parse_args(args)

1137

1138

def test_dump_json_conflicts_with_verbose(self):

1139

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

1140

with self.assertParseError():

1141

self.parse_args(args)

1142

1143

def test_dump_json_conflicts_with_action(self):

1144

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

1145

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

1146

with self.assertParseError():

1147

self.parse_args(args)

1148

1149

def test_all_can_not_be_alone(self):

1150

args = ["--all"]

1151

with self.assertParseError():

1152

self.parse_args(args)

1153

1154

def test_all_is_ok_with_any_action(self):

1155

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

1156

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

1157

self.parse_args(args)

1158

1159

def test_any_action_is_ok_with_one_client(self):

1160

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

1161

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

1162

self.parse_args(args)

1163

1164

def test_one_client_with_all_actions_except_is_enabled(self):

1165

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

1166

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

1167

continue

1168

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

1169

self.parse_args(args)

1170

1171

def test_two_clients_with_all_actions_except_is_enabled(self):

1172

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

1173

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

1174

continue

1175

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

1176

"client2")

1177

self.parse_args(args)

1178

1179

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1180

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

1181

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

1182

continue

1183

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

1184

"client2")

1185

self.parse_args(args)

1186

1187

def test_is_enabled_fails_without_client(self):

1188

args = ["--is-enabled"]

1189

with self.assertParseError():

1190

self.parse_args(args)

1191

1192

def test_is_enabled_fails_with_two_clients(self):

1193

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

1194

with self.assertParseError():

1195

self.parse_args(args)

1196

1197

def test_remove_can_only_be_combined_with_action_deny(self):

1198

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

1199

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

1200

continue

1201

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

1202

"--remove")

1203

with self.assertParseError():

1204

self.parse_args(args)

1205

1206

1207

class Test_dbus_exceptions(unittest.TestCase):

1208

1209

def test_dbus_ConnectFailed_is_Error(self):

1210

with self.assertRaises(dbus.Error):

1211

raise dbus.ConnectFailed()

1212

1213

1214

class Test_dbus_MandosBus(unittest.TestCase):

1215

1216

class MockMandosBus(dbus.MandosBus):

1217

def __init__(self):

1218

self._name = "se.recompile.Mandos"

1219

self._server_path = "/"

1220

self._server_interface = "se.recompile.Mandos"

1221

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

1222

self.calls = []

1223

self.call_method_return = Unique()

1224

1225

def call_method(self, methodname, busname, objectpath,

1226

interface, *args):

1227

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

1228

interface, args))

1229

return self.call_method_return

1230

1231

def setUp(self):

1232

self.bus = self.MockMandosBus()

1233

1234

def test_set_client_property(self):

1235

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

1236

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

1237

"org.freedesktop.DBus.Properties",

1238

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

1239

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

1240

1241

def test_call_client_method(self):

1242

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

1243

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

1244

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

1245

self.bus._client_interface, ())

1246

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

1247

1248

def test_call_client_method_with_args(self):

1249

args = (Unique(), Unique())

1250

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

1251

*args)

1252

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

1253

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

1254

self.bus._client_interface,

1255

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

1256

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

1257

1258

def test_get_clients_and_properties(self):

1259

managed_objects = {

1260

"objectpath": {

1261

self.bus._client_interface: {

1262

"key": "value",

1263

"bool": True,

1264

},

1265

"irrelevant_interface": {

1266

"key": "othervalue",

1267

"bool": False,

1268

},

1269

},

1270

"other_objectpath": {

1271

"other_irrelevant_interface": {

1272

"key": "value 3",

1273

"bool": None,

1274

},

1275

},

1276

}

1277

expected_clients_and_properties = {

1278

"objectpath": {

1279

"key": "value",

1280

"bool": True,

1281

}

1282

}

1283

self.bus.call_method_return = managed_objects

1284

ret = self.bus.get_clients_and_properties()

1285

self.assertDictEqual(expected_clients_and_properties, ret)

1286

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

1287

self.bus._server_path,

1288

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

1289

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

1290

1291

def test_call_server_method(self):

1292

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

1293

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

1294

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

1295

self.bus._server_path,

1296

self.bus._server_interface, ())

1297

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

1298

1299

def test_call_server_method_with_args(self):

1300

args = (Unique(), Unique())

1301

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

1302

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

1303

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

1304

self.bus._server_path,

1305

self.bus._server_interface,

1306

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

1307

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

1308

1309

1310

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1311

1312

def MockDBusPython_func(self, func):

1313

class mock_dbus_python(object):

1314

"""mock dbus-python module"""

1315

class exceptions(object):

1316

"""Pseudo-namespace"""

1317

class DBusException(Exception):

1318

pass

1319

class SystemBus(object):

1320

@staticmethod

1321

def get_object(busname, objectpath):

1322

DBusObject = collections.namedtuple(

1323

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

1324

def method(*args, **kwargs):

1325

self.assertEqual({"dbus_interface":

1326

"interface"},

1327

kwargs)

1328

return func(*args)

1329

def set_property(interface, key, value,

1330

dbus_interface=None):

1331

self.assertEqual(

1332

"org.freedesktop.DBus.Properties",

1333

dbus_interface)

1334

self.assertEqual("Secret", key)

1335

return func(interface, key, value,

1336

dbus_interface=dbus_interface)

1337

return DBusObject(methodname=method,

1338

Set=set_property)

1339

class Boolean(object):

1340

def __init__(self, value):

1341

self.value = bool(value)

1342

def __bool__(self):

1343

return self.value

1344

if sys.version_info.major == 2:

1345

__nonzero__ = __bool__

1346

class ObjectPath(str):

1347

pass

1348

class Dictionary(dict):

1349

pass

1350

class ByteArray(bytes):

1351

pass

1352

return mock_dbus_python

1353

1354

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

1355

interface, *args):

1356

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

1357

return bus.call_method(methodname, busname, objectpath,

1358

interface, *args)

1359

1360

def test_call_method_returns(self):

1361

expected_method_return = Unique()

1362

method_args = (Unique(), Unique())

1363

def func(*args):

1364

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

1365

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

1366

self.assertIs(marg, arg)

1367

return expected_method_return

1368

mock_dbus_python = self.MockDBusPython_func(func)

1369

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1370

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

1371

"objectpath", "interface",

1372

*method_args)

1373

self.assertIs(ret, expected_method_return)

1374

1375

def test_call_method_filters_bool_true(self):

1376

def func():

1377

return method_return

1378

mock_dbus_python = self.MockDBusPython_func(func)

1379

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1380

method_return = mock_dbus_python.Boolean(True)

1381

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

1382

"objectpath", "interface")

1383

self.assertTrue(ret)

1384

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1385

1386

def test_call_method_filters_bool_false(self):

1387

def func():

1388

return method_return

1389

mock_dbus_python = self.MockDBusPython_func(func)

1390

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1391

method_return = mock_dbus_python.Boolean(False)

1392

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

1393

"objectpath", "interface")

1394

self.assertFalse(ret)

1395

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1396

1397

def test_call_method_filters_objectpath(self):

1398

def func():

1399

return method_return

1400

mock_dbus_python = self.MockDBusPython_func(func)

1401

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1402

method_return = mock_dbus_python.ObjectPath("objectpath")

1403

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

1404

"objectpath", "interface")

1405

self.assertEqual("objectpath", ret)

1406

self.assertIsNot("objectpath", ret)

1407

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1408

1409

def test_call_method_filters_booleans_in_dict(self):

1410

def func():

1411

return method_return

1412

mock_dbus_python = self.MockDBusPython_func(func)

1413

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1414

method_return = mock_dbus_python.Dictionary(

1415

{mock_dbus_python.Boolean(True):

1416

mock_dbus_python.Boolean(False),

1417

mock_dbus_python.Boolean(False):

1418

mock_dbus_python.Boolean(True)})

1419

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

1420

"objectpath", "interface")

1421

expected_method_return = {True: False,

1422

False: True}

1423

self.assertEqual(expected_method_return, ret)

1424

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1425

1426

def test_call_method_filters_objectpaths_in_dict(self):

1427

def func():

1428

return method_return

1429

mock_dbus_python = self.MockDBusPython_func(func)

1430

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1431

method_return = mock_dbus_python.Dictionary(

1432

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1433

mock_dbus_python.ObjectPath("objectpath_value_1"),

1434

mock_dbus_python.ObjectPath("objectpath_key_2"):

1435

mock_dbus_python.ObjectPath("objectpath_value_2")})

1436

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

1437

"objectpath", "interface")

1438

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

1439

for key, value in

1440

method_return.items()}

1441

self.assertEqual(expected_method_return, ret)

1442

self.assertIsInstance(ret, dict)

1443

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1444

1445

def test_call_method_filters_dict_in_dict(self):

1446

def func():

1447

return method_return

1448

mock_dbus_python = self.MockDBusPython_func(func)

1449

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1450

method_return = mock_dbus_python.Dictionary(

1451

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

1452

"key12": "value12"}),

1453

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

1454

"key22": "value22"})})

1455

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

1456

"objectpath", "interface")

1457

expected_method_return = {

1458

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

1459

"key12": "value12"},

1460

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

1461

"key22": "value22"},

1462

}

1463

self.assertEqual(expected_method_return, ret)

1464

self.assertIsInstance(ret, dict)

1465

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1466

for key, value in ret.items():

1467

self.assertIsInstance(value, dict)

1468

self.assertEqual(expected_method_return[key], value)

1469

self.assertNotIsInstance(value,

1470

mock_dbus_python.Dictionary)

1471

1472

def test_call_method_filters_dict_three_deep(self):

1473

def func():

1474

return method_return

1475

mock_dbus_python = self.MockDBusPython_func(func)

1476

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1477

method_return = mock_dbus_python.Dictionary(

1478

{"key1":

1479

mock_dbus_python.Dictionary(

1480

{"key2":

1481

mock_dbus_python.Dictionary(

1482

{"key3":

1483

mock_dbus_python.Boolean(True),

1484

}),

1485

}),

1486

})

1487

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

1488

"objectpath", "interface")

1489

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

1490

self.assertEqual(expected_method_return, ret)

1491

self.assertIsInstance(ret, dict)

1492

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1493

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

1494

self.assertNotIsInstance(ret["key1"],

1495

mock_dbus_python.Dictionary)

1496

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

1497

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

1498

mock_dbus_python.Dictionary)

1499

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

1500

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

1501

mock_dbus_python.Boolean)

1502

1503

def test_call_method_handles_exception(self):

1504

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

1505

1506

def func():

1507

dbus_logger.error("Test")

1508

raise mock_dbus_python.exceptions.DBusException()

1509

1510

mock_dbus_python = self.MockDBusPython_func(func)

1511

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1512

1513

class CountingHandler(logging.Handler):

1514

count = 0

1515

def emit(self, record):

1516

self.count += 1

1517

1518

counting_handler = CountingHandler()

1519

1520

dbus_logger.addHandler(counting_handler)

1521

1522

try:

1523

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

1524

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

1525

"objectpath", "interface")

1526

finally:

1527

dbus_logger.removeFilter(counting_handler)

1528

1529

self.assertNotIsInstance(e, dbus.ConnectFailed)

1530

1531

# Make sure the dbus logger was suppressed

1532

self.assertEqual(0, counting_handler.count)

1533

1534

def test_Set_Secret_sends_bytearray(self):

1535

ret = [None]

1536

def func(*args, **kwargs):

1537

ret[0] = (args, kwargs)

1538

mock_dbus_python = self.MockDBusPython_func(func)

1539

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1540

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

1541

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

1542

mock_dbus_python.ByteArray(b"value")),

1543

{"dbus_interface":

1544

"org.freedesktop.DBus.Properties"})

1545

self.assertEqual(expected_call, ret[0])

1546

if sys.version_info.major == 2:

1547

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

1548

mock_dbus_python.ByteArray)

1549

1550

def test_get_object_converts_to_correct_exception(self):

1551

bus = dbus_python_adapter.SystemBus(

1552

self.fake_dbus_python_raises_exception_on_connect)

1553

with self.assertRaises(dbus.ConnectFailed):

1554

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

1555

"objectpath", "interface")

1556

1557

class fake_dbus_python_raises_exception_on_connect(object):

1558

"""fake dbus-python module"""

1559

class exceptions(object):

1560

"""Pseudo-namespace"""

1561

class DBusException(Exception):

1562

pass

1563

1564

@classmethod

1565

def SystemBus(cls):

1566

def get_object(busname, objectpath):

1567

raise cls.exceptions.DBusException()

1568

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

1569

return Bus(get_object=get_object)

1570

1571

1572

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1573

class mock_dbus_python(object):

1574

"""mock dbus-python modules"""

1575

class SystemBus(object):

1576

@staticmethod

1577

def get_object(busname, objectpath):

1578

return Unique()

1579

1580

def setUp(self):

1581

self.bus = dbus_python_adapter.CachingBus(

1582

self.mock_dbus_python)

1583

1584

def test_returns_distinct_objectpaths(self):

1585

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

1586

self.assertIsInstance(obj1, Unique)

1587

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

1588

self.assertIsInstance(obj2, Unique)

1589

self.assertIsNot(obj1, obj2)

1590

1591

def test_returns_distinct_busnames(self):

1592

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

1593

self.assertIsInstance(obj1, Unique)

1594

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

1595

self.assertIsInstance(obj2, Unique)

1596

self.assertIsNot(obj1, obj2)

1597

1598

def test_returns_distinct_both(self):

1599

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

1600

self.assertIsInstance(obj1, Unique)

1601

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

1602

self.assertIsInstance(obj2, Unique)

1603

self.assertIsNot(obj1, obj2)

1604

1605

def test_returns_same(self):

1606

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

1607

self.assertIsInstance(obj1, Unique)

1608

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

1609

self.assertIsInstance(obj2, Unique)

1610

self.assertIs(obj1, obj2)

1611

1612

def test_returns_same_old(self):

1613

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

1614

self.assertIsInstance(obj1, Unique)

1615

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

1616

self.assertIsInstance(obj2, Unique)

1617

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

1618

self.assertIsInstance(obj1b, Unique)

1619

self.assertIsNot(obj1, obj2)

1620

self.assertIsNot(obj2, obj1b)

1621

self.assertIs(obj1, obj1b)

1622

1623

1624

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1625

1626

def Stub_pydbus_func(self, func):

1627

class stub_pydbus(object):

1628

"""stub pydbus module"""

1629

class SystemBus(object):

1630

@staticmethod

1631

def get(busname, objectpath):

1632

DBusObject = collections.namedtuple(

1633

"DBusObject", ("methodname",))

1634

return {"interface":

1635

DBusObject(methodname=func)}

1636

return stub_pydbus

1637

1638

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

1639

interface, *args):

1640

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

1641

return bus.call_method(methodname, busname, objectpath,

1642

interface, *args)

1643

1644

def test_call_method_returns(self):

1645

expected_method_return = Unique()

1646

method_args = (Unique(), Unique())

1647

def func(*args):

1648

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

1649

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

1650

self.assertIs(marg, arg)

1651

return expected_method_return

1652

stub_pydbus = self.Stub_pydbus_func(func)

1653

bus = pydbus_adapter.SystemBus(stub_pydbus)

1654

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

1655

"objectpath", "interface",

1656

*method_args)

1657

self.assertIs(ret, expected_method_return)

1658

1659

def test_call_method_handles_exception(self):

1660

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

1661

1662

def func():

1663

raise gi.repository.GLib.Error()

1664

1665

stub_pydbus = self.Stub_pydbus_func(func)

1666

bus = pydbus_adapter.SystemBus(stub_pydbus)

1667

1668

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

1669

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

1670

"objectpath", "interface")

1671

1672

self.assertNotIsInstance(e, dbus.ConnectFailed)

1673

1674

def test_get_converts_to_correct_exception(self):

1675

bus = pydbus_adapter.SystemBus(

1676

self.fake_pydbus_raises_exception_on_connect)

1677

with self.assertRaises(dbus.ConnectFailed):

1678

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

1679

"objectpath", "interface")

1680

1681

class fake_pydbus_raises_exception_on_connect(object):

1682

"""fake dbus-python module"""

1683

@classmethod

1684

def SystemBus(cls):

1685

def get(busname, objectpath):

1686

raise gi.repository.GLib.Error()

1687

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

1688

return Bus(get=get)

1689

1690

def test_set_property_uses_setattr(self):

1691

class Object(object):

1692

pass

1693

obj = Object()

1694

class pydbus_spy(object):

1695

class SystemBus(object):

1696

@staticmethod

1697

def get(busname, objectpath):

1698

return {"interface": obj}

1699

bus = pydbus_adapter.SystemBus(pydbus_spy)

1700

value = Unique()

1701

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

1702

value)

1703

self.assertIs(value, obj.key)

1704

1705

def test_get_suppresses_xml_deprecation_warning(self):

1706

if sys.version_info.major >= 3:

1707

return

1708

class stub_pydbus_get(object):

1709

class SystemBus(object):

1710

@staticmethod

1711

def get(busname, objectpath):

1712

warnings.warn_explicit(

1713

"deprecated", DeprecationWarning,

1714

"xml.etree.ElementTree", 0)

1715

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1716

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

1717

warnings.simplefilter("always")

1718

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

1719

self.assertEqual(0, len(w))

1720

1721

1722

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1723

class stub_pydbus(object):

1724

"""stub pydbus module"""

1725

class SystemBus(object):

1726

@staticmethod

1727

def get(busname, objectpath):

1728

return Unique()

1729

1730

def setUp(self):

1731

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

1732

1733

def test_returns_distinct_objectpaths(self):

1734

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

1735

self.assertIsInstance(obj1, Unique)

1736

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

1737

self.assertIsInstance(obj2, Unique)

1738

self.assertIsNot(obj1, obj2)

1739

1740

def test_returns_distinct_busnames(self):

1741

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

1742

self.assertIsInstance(obj1, Unique)

1743

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

1744

self.assertIsInstance(obj2, Unique)

1745

self.assertIsNot(obj1, obj2)

1746

1747

def test_returns_distinct_both(self):

1748

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

1749

self.assertIsInstance(obj1, Unique)

1750

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

1751

self.assertIsInstance(obj2, Unique)

1752

self.assertIsNot(obj1, obj2)

1753

1754

def test_returns_same(self):

1755

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

1756

self.assertIsInstance(obj1, Unique)

1757

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

1758

self.assertIsInstance(obj2, Unique)

1759

self.assertIs(obj1, obj2)

1760

1761

def test_returns_same_old(self):

1762

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

1763

self.assertIsInstance(obj1, Unique)

1764

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

1765

self.assertIsInstance(obj2, Unique)

1766

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

1767

self.assertIsInstance(obj1b, Unique)

1768

self.assertIsNot(obj1, obj2)

1769

self.assertIsNot(obj2, obj1b)

1770

self.assertIs(obj1, obj1b)

1771

1772

1773

class Test_commands_from_options(unittest.TestCase):

1774

1775

def setUp(self):

1776

self.parser = argparse.ArgumentParser()

1777

add_command_line_options(self.parser)

1778

1779

def test_is_enabled(self):

1780

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

1781

command.IsEnabled)

1782

1783

def assert_command_from_args(self, args, command_cls,

1784

**cmd_attrs):

1785

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

1786

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

1787

options = self.parser.parse_args(args)

1788

check_option_syntax(self.parser, options)

1789

commands = commands_from_options(options)

1790

self.assertEqual(1, len(commands))

1791

command = commands[0]

1792

self.assertIsInstance(command, command_cls)

1793

for key, value in cmd_attrs.items():

1794

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

1795

1796

def test_is_enabled_short(self):

1797

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

1798

command.IsEnabled)

1799

1800

def test_approve(self):

1801

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

1802

command.Approve)

1803

1804

def test_approve_short(self):

1805

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

1806

command.Approve)

1807

1808

def test_deny(self):

1809

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

1810

command.Deny)

1811

1812

def test_deny_short(self):

1813

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

1814

1815

def test_remove(self):

1816

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

1817

command.Remove)

1818

1819

def test_deny_before_remove(self):

1820

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

1821

"client"])

1822

check_option_syntax(self.parser, options)

1823

commands = commands_from_options(options)

1824

self.assertEqual(2, len(commands))

1825

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

1826

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

1827

1828

def test_deny_before_remove_reversed(self):

1829

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

1830

"--all"])

1831

check_option_syntax(self.parser, options)

1832

commands = commands_from_options(options)

1833

self.assertEqual(2, len(commands))

1834

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

1835

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

1836

1837

def test_remove_short(self):

1838

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

1839

command.Remove)

1840

1841

def test_dump_json(self):

1842

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

1843

command.DumpJSON)

1844

1845

def test_enable(self):

1846

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

1847

command.Enable)

1848

1849

def test_enable_short(self):

1850

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

1851

command.Enable)

1852

1853

def test_disable(self):

1854

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

1855

command.Disable)

1856

1857

def test_disable_short(self):

1858

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

1859

command.Disable)

1860

1861

def test_bump_timeout(self):

1862

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

1863

command.BumpTimeout)

1864

1865

def test_bump_timeout_short(self):

1866

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

1867

command.BumpTimeout)

1868

1869

def test_start_checker(self):

1870

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

1871

command.StartChecker)

1872

1873

def test_stop_checker(self):

1874

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

1875

command.StopChecker)

1876

1877

def test_approve_by_default(self):

1878

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

1879

"client"],

1880

command.ApproveByDefault)

1881

1882

def test_deny_by_default(self):

1883

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

1884

command.DenyByDefault)

1885

1886

def test_checker(self):

1887

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

1888

command.SetChecker,

1889

value_to_set=":")

1890

1891

def test_checker_empty(self):

1892

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

1893

command.SetChecker,

1894

value_to_set="")

1895

1896

def test_checker_short(self):

1897

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

1898

command.SetChecker,

1899

value_to_set=":")

1900

1901

def test_host(self):

1902

self.assert_command_from_args(

1903

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

1904

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

1905

1906

def test_host_short(self):

1907

self.assert_command_from_args(

1908

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

1909

value_to_set="client.example.org")

1910

1911

def test_secret_devnull(self):

1912

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

1913

"client"], command.SetSecret,

1914

value_to_set=b"")

1915

1916

def test_secret_tempfile(self):

1917

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

1918

value = b"secret\0xyzzy\nbar"

1919

f.write(value)

1920

f.seek(0)

1921

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

1922

"client"],

1923

command.SetSecret,

1924

value_to_set=value)

1925

1926

def test_secret_devnull_short(self):

1927

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

1928

"client"], command.SetSecret,

1929

value_to_set=b"")

1930

1931

def test_secret_tempfile_short(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(["-s", f.name, "client"],

1937

command.SetSecret,

1938

value_to_set=value)

1939

1940

def test_timeout(self):

1941

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

1942

command.SetTimeout,

1943

value_to_set=300000)

1944

1945

def test_timeout_short(self):

1946

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

1947

command.SetTimeout,

1948

value_to_set=300000)

1949

1950

def test_extended_timeout(self):

1951

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

1952

"client"],

1953

command.SetExtendedTimeout,

1954

value_to_set=900000)

1955

1956

def test_interval(self):

1957

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

1958

"client"], command.SetInterval,

1959

value_to_set=120000)

1960

1961

def test_interval_short(self):

1962

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

1963

command.SetInterval,

1964

value_to_set=120000)

1965

1966

def test_approval_delay(self):

1967

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

1968

"client"],

1969

command.SetApprovalDelay,

1970

value_to_set=30000)

1971

1972

def test_approval_duration(self):

1973

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

1974

"client"],

1975

command.SetApprovalDuration,

1976

value_to_set=1000)

1977

1978

def test_print_table(self):

1979

self.assert_command_from_args([], command.PrintTable,

1980

verbose=False)

1981

1982

def test_print_table_verbose(self):

1983

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

1984

command.PrintTable,

1985

verbose=True)

1986

1987

def test_print_table_verbose_short(self):

1988

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

1989

verbose=True)

1990

1991

1992

class TestCommand(unittest.TestCase):

1993

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

1994

1995

class FakeMandosBus(dbus.MandosBus):

1996

def __init__(self, testcase):

1997

self.client_properties = {

1998

"Name": "foo",

1999

"KeyID": ("92ed150794387c03ce684574b1139a65"

2000

"94a34f895daaaf09fd8ea90a27cddb12"),

2001

"Secret": b"secret",

2002

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

2003

"Enabled": True,

2004

"Timeout": 300000,

2005

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

2006

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

2007

"Interval": 120000,

2008

"Fingerprint": ("778827225BA7DE539C5A"

2009

"7CFA59CFF7CDBD9A5920"),

2010

"CheckerRunning": False,

2011

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

2012

"ApprovalPending": False,

2013

"ApprovedByDefault": True,

2014

"LastApprovalRequest": "",

2015

"ApprovalDelay": 0,

2016

"ApprovalDuration": 1000,

2017

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

2018

"ExtendedTimeout": 900000,

2019

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

2020

"LastCheckerStatus": 0,

2021

}

2022

self.other_client_properties = {

2023

"Name": "barbar",

2024

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2025

"6ab612cff5ad227247e46c2b020f441c"),

2026

"Secret": b"secretbar",

2027

"Host": "192.0.2.3",

2028

"Enabled": True,

2029

"Timeout": 300000,

2030

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

2031

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

2032

"Interval": 120000,

2033

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2034

"F547B3A107558FCA3A27"),

2035

"CheckerRunning": True,

2036

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

2037

"ApprovalPending": False,

2038

"ApprovedByDefault": False,

2039

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

2040

"ApprovalDelay": 30000,

2041

"ApprovalDuration": 93785000,

2042

"Checker": ":",

2043

"ExtendedTimeout": 900000,

2044

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

2045

"LastCheckerStatus": -2,

2046

}

2047

self.clients = collections.OrderedDict(

2048

[

2049

("client_objectpath", self.client_properties),

2050

("other_client_objectpath",

2051

self.other_client_properties),

2052

])

2053

self.one_client = {"client_objectpath":

2054

self.client_properties}

2055

self.testcase = testcase

2056

self.calls = []

2057

2058

def call_method(self, methodname, busname, objectpath,

2059

interface, *args):

2060

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

2061

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

2062

interface, args))

2063

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

2064

if methodname == "Set":

2065

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

2066

interface, key, value = args

2067

self.testcase.assertEqual(

2068

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

2069

self.clients[objectpath][key] = value

2070

return

2071

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

2072

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

2073

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

2074

clientpath = args[0]

2075

del self.clients[clientpath]

2076

return

2077

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

2078

if methodname == "Approve":

2079

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

2080

return

2081

raise ValueError()

2082

2083

def setUp(self):

2084

self.bus = self.FakeMandosBus(self)

2085

2086

2087

class TestBaseCommands(TestCommand):

2088

2089

def test_IsEnabled_exits_successfully(self):

2090

with self.assertRaises(SystemExit) as e:

2091

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

2092

if e.exception.code is not None:

2093

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

2094

else:

2095

self.assertIsNone(e.exception.code)

2096

2097

def test_IsEnabled_exits_with_failure(self):

2098

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

2099

with self.assertRaises(SystemExit) as e:

2100

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

2101

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

2102

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

2103

else:

2104

self.assertIsNotNone(e.exception.code)

2105

2106

def test_Approve(self):

2107

busname = "se.recompile.Mandos"

2108

client_interface = "se.recompile.Mandos.Client"

2109

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

2110

for clientpath in self.bus.clients:

2111

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

2112

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

2113

2114

def test_Deny(self):

2115

busname = "se.recompile.Mandos"

2116

client_interface = "se.recompile.Mandos.Client"

2117

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

2118

for clientpath in self.bus.clients:

2119

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

2120

client_interface, (False,)),

2121

self.bus.calls)

2122

2123

def test_Remove(self):

2124

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

2125

for clientpath in self.bus.clients:

2126

self.assertIn(("RemoveClient", dbus_busname,

2127

dbus_server_path, dbus_server_interface,

2128

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

2129

2130

expected_json = {

2131

"foo": {

2132

"Name": "foo",

2133

"KeyID": ("92ed150794387c03ce684574b1139a65"

2134

"94a34f895daaaf09fd8ea90a27cddb12"),

2135

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

2136

"Enabled": True,

2137

"Timeout": 300000,

2138

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

2139

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

2140

"Interval": 120000,

2141

"Fingerprint": ("778827225BA7DE539C5A"

2142

"7CFA59CFF7CDBD9A5920"),

2143

"CheckerRunning": False,

2144

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

2145

"ApprovalPending": False,

2146

"ApprovedByDefault": True,

2147

"LastApprovalRequest": "",

2148

"ApprovalDelay": 0,

2149

"ApprovalDuration": 1000,

2150

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

2151

"ExtendedTimeout": 900000,

2152

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

2153

"LastCheckerStatus": 0,

2154

},

2155

"barbar": {

2156

"Name": "barbar",

2157

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2158

"6ab612cff5ad227247e46c2b020f441c"),

2159

"Host": "192.0.2.3",

2160

"Enabled": True,

2161

"Timeout": 300000,

2162

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

2163

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

2164

"Interval": 120000,

2165

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2166

"F547B3A107558FCA3A27"),

2167

"CheckerRunning": True,

2168

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

2169

"ApprovalPending": False,

2170

"ApprovedByDefault": False,

2171

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

2172

"ApprovalDelay": 30000,

2173

"ApprovalDuration": 93785000,

2174

"Checker": ":",

2175

"ExtendedTimeout": 900000,

2176

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

2177

"LastCheckerStatus": -2,

2178

},

2179

}

2180

2181

def test_DumpJSON_normal(self):

2182

with self.capture_stdout_to_buffer() as buffer:

2183

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

2184

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

2185

self.assertDictEqual(self.expected_json, json_data)

2186

2187

@staticmethod

2188

@contextlib.contextmanager

2189

def capture_stdout_to_buffer():

2190

capture_buffer = io.StringIO()

2191

old_stdout = sys.stdout

2192

sys.stdout = capture_buffer

2193

try:

2194

yield capture_buffer

2195

finally:

2196

sys.stdout = old_stdout

2197

2198

def test_DumpJSON_one_client(self):

2199

with self.capture_stdout_to_buffer() as buffer:

2200

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

2201

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

2202

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

2203

self.assertDictEqual(expected_json, json_data)

2204

2205

def test_PrintTable_normal(self):

2206

with self.capture_stdout_to_buffer() as buffer:

2207

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

2208

expected_output = "\n".join((

2209

"Name Enabled Timeout Last Successful Check",

2210

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

2211

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

2212

)) + "\n"

2213

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

2214

2215

def test_PrintTable_verbose(self):

2216

with self.capture_stdout_to_buffer() as buffer:

2217

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

2218

columns = (

2219

(

2220

"Name ",

2221

"foo ",

2222

"barbar ",

2223

),(

2224

"Enabled ",

2225

"Yes ",

2226

"Yes ",

2227

),(

2228

"Timeout ",

2229

"00:05:00 ",

2230

"00:05:00 ",

2231

),(

2232

"Last Successful Check ",

2233

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

2234

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

2235

),(

2236

"Created ",

2237

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

2238

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

2239

),(

2240

"Interval ",

2241

"00:02:00 ",

2242

"00:02:00 ",

2243

),(

2244

"Host ",

2245

"foo.example.org ",

2246

"192.0.2.3 ",

2247

),(

2248

("Key ID "

2249

" "),

2250

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2251

"ea90a27cddb12 "),

2252

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2253

"46c2b020f441c "),

2254

),(

2255

"Fingerprint ",

2256

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2257

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2258

),(

2259

"Check Is Running ",

2260

"No ",

2261

"Yes ",

2262

),(

2263

"Last Enabled ",

2264

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

2265

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

2266

),(

2267

"Approval Is Pending ",

2268

"No ",

2269

"No ",

2270

),(

2271

"Approved By Default ",

2272

"Yes ",

2273

"No ",

2274

),(

2275

"Last Approval Request ",

2276

" ",

2277

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

2278

),(

2279

"Approval Delay ",

2280

"00:00:00 ",

2281

"00:00:30 ",

2282

),(

2283

"Approval Duration ",

2284

"00:00:01 ",

2285

"1T02:03:05 ",

2286

),(

2287

"Checker ",

2288

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

2289

": ",

2290

),(

2291

"Extended Timeout ",

2292

"00:15:00 ",

2293

"00:15:00 ",

2294

),(

2295

"Expires ",

2296

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

2297

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

2298

),(

2299

"Last Checker Status",

2300

"0 ",

2301

"-2 ",

2302

)

2303

)

2304

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

2305

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

2306

for rows in columns)

2307

for line in range(num_lines))

2308

+ "\n")

2309

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

2310

2311

def test_PrintTable_one_client(self):

2312

with self.capture_stdout_to_buffer() as buffer:

2313

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

2314

expected_output = "\n".join((

2315

"Name Enabled Timeout Last Successful Check",

2316

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

2317

)) + "\n"

2318

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

2319

2320

2321

class TestPropertySetterCmd(TestCommand):

2322

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

2323

2324

def runTest(self):

2325

if not hasattr(self, "command"):

2326

return # Abstract TestCase class

2327

2328

if hasattr(self, "values_to_set"):

2329

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

2330

values_to_get = getattr(self, "values_to_get",

2331

self.values_to_set)

2332

else:

2333

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

2334

values_to_get = self.values_to_get

2335

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

2336

for clientpath in self.bus.clients:

2337

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

2338

Unique())

2339

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

2340

for clientpath in self.bus.clients:

2341

value = (self.bus.clients[clientpath]

2342

[self.propname])

2343

self.assertNotIsInstance(value, Unique)

2344

self.assertEqual(value_to_get, value)

2345

2346

2347

class TestEnableCmd(TestPropertySetterCmd):

2348

command = command.Enable

2349

propname = "Enabled"

2350

values_to_get = [True]

2351

2352

2353

class TestDisableCmd(TestPropertySetterCmd):

2354

command = command.Disable

2355

propname = "Enabled"

2356

values_to_get = [False]

2357

2358

2359

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2360

command = command.BumpTimeout

2361

propname = "LastCheckedOK"

2362

values_to_get = [""]

2363

2364

2365

class TestStartCheckerCmd(TestPropertySetterCmd):

2366

command = command.StartChecker

2367

propname = "CheckerRunning"

2368

values_to_get = [True]

2369

2370

2371

class TestStopCheckerCmd(TestPropertySetterCmd):

2372

command = command.StopChecker

2373

propname = "CheckerRunning"

2374

values_to_get = [False]

2375

2376

2377

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2378

command = command.ApproveByDefault

2379

propname = "ApprovedByDefault"

2380

values_to_get = [True]

2381

2382

2383

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2384

command = command.DenyByDefault

2385

propname = "ApprovedByDefault"

2386

values_to_get = [False]

2387

2388

2389

class TestSetCheckerCmd(TestPropertySetterCmd):

2390

command = command.SetChecker

2391

propname = "Checker"

2392

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

2393

2394

2395

class TestSetHostCmd(TestPropertySetterCmd):

2396

command = command.SetHost

2397

propname = "Host"

2398

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

2399

2400

2401

class TestSetSecretCmd(TestPropertySetterCmd):

2402

command = command.SetSecret

2403

propname = "Secret"

2404

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

2405

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

2406

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

2407

2408

2409

class TestSetTimeoutCmd(TestPropertySetterCmd):

2410

command = command.SetTimeout

2411

propname = "Timeout"

2412

values_to_set = [datetime.timedelta(),

2413

datetime.timedelta(minutes=5),

2414

datetime.timedelta(seconds=1),

2415

datetime.timedelta(weeks=1),

2416

datetime.timedelta(weeks=52)]

2417

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

2418

2419

2420

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2421

command = command.SetExtendedTimeout

2422

propname = "ExtendedTimeout"

2423

values_to_set = [datetime.timedelta(),

2424

datetime.timedelta(minutes=5),

2425

datetime.timedelta(seconds=1),

2426

datetime.timedelta(weeks=1),

2427

datetime.timedelta(weeks=52)]

2428

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

2429

2430

2431

class TestSetIntervalCmd(TestPropertySetterCmd):

2432

command = command.SetInterval

2433

propname = "Interval"

2434

values_to_set = [datetime.timedelta(),

2435

datetime.timedelta(minutes=5),

2436

datetime.timedelta(seconds=1),

2437

datetime.timedelta(weeks=1),

2438

datetime.timedelta(weeks=52)]

2439

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

2440

2441

2442

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2443

command = command.SetApprovalDelay

2444

propname = "ApprovalDelay"

2445

values_to_set = [datetime.timedelta(),

2446

datetime.timedelta(minutes=5),

2447

datetime.timedelta(seconds=1),

2448

datetime.timedelta(weeks=1),

2449

datetime.timedelta(weeks=52)]

2450

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

2451

2452

2453

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2454

command = command.SetApprovalDuration

2455

propname = "ApprovalDuration"

2456

values_to_set = [datetime.timedelta(),

2457

datetime.timedelta(minutes=5),

2458

datetime.timedelta(seconds=1),

2459

datetime.timedelta(weeks=1),

2460

datetime.timedelta(weeks=52)]

2461

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

2462

2463

2464

2465

def should_only_run_tests():

2466

parser = argparse.ArgumentParser(add_help=False)

2467

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

2468

args, unknown_args = parser.parse_known_args()

2469

run_tests = args.check

2470

if run_tests:

2471

# Remove --check argument from sys.argv

2472

sys.argv[1:] = unknown_args

2473

return run_tests

2474

2475

# Add all tests from doctest strings

2476

def load_tests(loader, tests, none):

2477

import doctest

2478

tests.addTests(doctest.DocTestSuite())

2479

return tests

2480

2481

if __name__ == "__main__":

2482

try:

2483

if should_only_run_tests():

2484

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

2485

unittest.main()

2486

else:

2487

main()

2488

finally:

2489

logging.shutdown()