/mandos/trunk : revision 423

1

#!/usr/bin/python

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

2

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

3

4

from __future__ import division

35

5

import sys

36

import argparse

6

import dbus

7

from optparse import OptionParser

37

8

import locale

38

9

import datetime

39

10

import re

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

48

import dbus

49

50

# Show warnings by default

51

if not sys.warnoptions:

52

import warnings

53

warnings.simplefilter("default")

54

55

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

56

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

57

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

58

59

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

60

61

if sys.version_info.major == 2:

62

str = unicode

63

64

locale.setlocale(locale.LC_ALL, "")

65

66

domain = "se.recompile"

67

busname = domain + ".Mandos"

68

server_path = "/"

69

server_interface = domain + ".Mandos"

70

client_interface = domain + ".Mandos.Client"

71

version = "1.8.3"

72

73

11

12

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

13

14

tablewords = {

15

'Name': u'Name',

16

'Enabled': u'Enabled',

17

'Timeout': u'Timeout',

18

'LastCheckedOK': u'Last Successful Check',

19

'Created': u'Created',

20

'Interval': u'Interval',

21

'Host': u'Host',

22

'Fingerprint': u'Fingerprint',

23

'CheckerRunning': u'Check Is Running',

24

'LastEnabled': u'Last Enabled',

25

'Checker': u'Checker',

26

}

27

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

28

domain = 'se.bsnet.fukt'

29

busname = domain + '.Mandos'

30

server_path = '/'

31

server_interface = domain + '.Mandos'

32

client_interface = domain + '.Mandos.Client'

33

version = "1.0.14"

74

34

try:

75

dbus.OBJECT_MANAGER_IFACE

76

except AttributeError:

77

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

35

bus = dbus.SystemBus()

36

mandos_dbus_objc = bus.get_object(busname, server_path)

37

except dbus.exceptions.DBusException:

38

sys.exit(1)

39

40

mandos_serv = dbus.Interface(mandos_dbus_objc,

41

dbus_interface = server_interface)

42

mandos_clients = mandos_serv.GetAllClientsWithProperties()

78

43

44

def timedelta_to_milliseconds(td):

45

"Convert a datetime.timedelta object to milliseconds"

46

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

47

+ (td.seconds * 1000)

48

+ (td.microseconds // 1000))

79

49

80

50

def milliseconds_to_string(ms):

81

51

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

82

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

83

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

84

hours=td.seconds // 3600,

85

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

86

seconds=td.seconds % 60))

87

88

89

def rfc3339_duration_to_delta(duration):

90

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

91

92

>>> rfc3339_duration_to_delta("P7D")

93

datetime.timedelta(7)

94

>>> rfc3339_duration_to_delta("PT60S")

95

datetime.timedelta(0, 60)

96

>>> rfc3339_duration_to_delta("PT60M")

97

datetime.timedelta(0, 3600)

98

>>> rfc3339_duration_to_delta("P60M")

99

datetime.timedelta(1680)

100

>>> rfc3339_duration_to_delta("PT24H")

101

datetime.timedelta(1)

102

>>> rfc3339_duration_to_delta("P1W")

103

datetime.timedelta(7)

104

>>> rfc3339_duration_to_delta("PT5M30S")

105

datetime.timedelta(0, 330)

106

>>> rfc3339_duration_to_delta("P1DT3M20S")

107

datetime.timedelta(1, 200)

108

>>> # Can not be empty:

109

>>> rfc3339_duration_to_delta("")

110

Traceback (most recent call last):

111

...

112

ValueError: Invalid RFC 3339 duration: u''

113

>>> # Must start with "P":

114

>>> rfc3339_duration_to_delta("1D")

115

Traceback (most recent call last):

116

...

117

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

118

>>> # Must use correct order

119

>>> rfc3339_duration_to_delta("PT1S2M")

120

Traceback (most recent call last):

121

...

122

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

123

>>> # Time needs time marker

124

>>> rfc3339_duration_to_delta("P1H2S")

125

Traceback (most recent call last):

126

...

127

ValueError: Invalid RFC 3339 duration: u'P1H2S'

128

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

129

>>> rfc3339_duration_to_delta("P1D2W")

130

Traceback (most recent call last):

131

...

132

ValueError: Invalid RFC 3339 duration: u'P1D2W'

133

>>> rfc3339_duration_to_delta("P2W2H")

134

Traceback (most recent call last):

135

...

136

ValueError: Invalid RFC 3339 duration: u'P2W2H'

137

"""

138

139

# Parsing an RFC 3339 duration with regular expressions is not

140

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

141

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

142

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

143

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

144

# avoid excessive use of external libraries.

145

146

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

147

Token = collections.namedtuple("Token", (

148

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

149

# a "group" containing digits

150

"value", # datetime.timedelta or None

151

"followers")) # Tokens valid after this token

152

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

153

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

154

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

155

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

156

datetime.timedelta(seconds=1),

157

frozenset((token_end, )))

158

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

159

datetime.timedelta(minutes=1),

160

frozenset((token_second, token_end)))

161

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

162

datetime.timedelta(hours=1),

163

frozenset((token_minute, token_end)))

164

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

165

None,

166

frozenset((token_hour, token_minute,

167

token_second)))

168

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

169

datetime.timedelta(days=1),

170

frozenset((token_time, token_end)))

171

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

172

datetime.timedelta(weeks=4),

173

frozenset((token_day, token_end)))

174

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

175

datetime.timedelta(weeks=52),

176

frozenset((token_month, token_end)))

177

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

178

datetime.timedelta(weeks=1),

179

frozenset((token_end, )))

180

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

181

frozenset((token_year, token_month,

182

token_day, token_time,

183

token_week)))

184

# Define starting values:

185

# Value so far

186

value = datetime.timedelta()

187

found_token = None

188

# Following valid tokens

189

followers = frozenset((token_duration, ))

190

# String left to parse

191

s = duration

192

# Loop until end token is found

193

while found_token is not token_end:

194

# Search for any currently valid tokens

195

for token in followers:

196

match = token.regexp.match(s)

197

if match is not None:

198

# Token found

199

if token.value is not None:

200

# Value found, parse digits

201

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

202

# Add to value so far

203

value += factor * token.value

204

# Strip token from string

205

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

206

# Go to found token

207

found_token = token

208

# Set valid next tokens

209

followers = found_token.followers

210

break

211

else:

212

# No currently valid tokens were found

213

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

214

.format(duration))

215

# End token found

216

return value

52

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

53

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

54

"hours": td.seconds // 3600,

55

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

56

"seconds": td.seconds % 60,

57

})

217

58

218

59

219

60

def string_to_delta(interval):

220

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

221

222

try:

223

return rfc3339_duration_to_delta(interval)

224

except ValueError as e:

225

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

226

' '.join(e.args))

227

return parse_pre_1_6_1_interval(interval)

228

229

230

def parse_pre_1_6_1_interval(interval):

231

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

232

and return a datetime.timedelta

233

234

>>> parse_pre_1_6_1_interval('7d')

61

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

62

63

>>> string_to_delta('7d')

235

64

datetime.timedelta(7)

236

>>> parse_pre_1_6_1_interval('60s')

65

>>> string_to_delta('60s')

237

66

datetime.timedelta(0, 60)

238

>>> parse_pre_1_6_1_interval('60m')

67

>>> string_to_delta('60m')

239

68

datetime.timedelta(0, 3600)

240

>>> parse_pre_1_6_1_interval('24h')

69

>>> string_to_delta('24h')

241

70

datetime.timedelta(1)

242

>>> parse_pre_1_6_1_interval('1w')

71

>>> string_to_delta(u'1w')

243

72

datetime.timedelta(7)

244

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

73

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

245

74

datetime.timedelta(0, 330)

246

>>> parse_pre_1_6_1_interval('')

247

datetime.timedelta(0)

248

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

249

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

250

datetime.timedelta(2, 480, 18000)

251

252

75

"""

253

254

value = datetime.timedelta(0)

255

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

256

257

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

258

if suffix == "d":

259

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

260

elif suffix == "s":

261

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

262

elif suffix == "m":

263

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

264

elif suffix == "h":

265

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

266

elif suffix == "w":

267

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

268

elif suffix == "":

269

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

270

return value

271

272

273

## Classes for commands.

274

275

# Abstract classes first

276

class Command(object):

277

"""Abstract class for commands"""

278

def run(self, mandos, clients):

279

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

280

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

281

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

282

self.mandos = mandos

283

for client, properties in clients.items():

284

self.run_on_one_client(client, properties)

285

286

class PrintCmd(Command):

287

"""Abstract class for commands printing client details"""

288

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

289

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

290

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

291

"ApprovalPending", "ApprovedByDefault",

292

"LastApprovalRequest", "ApprovalDelay",

293

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

294

"Expires", "LastCheckerStatus")

295

def run(self, mandos, clients):

296

print(self.output(clients))

297

298

class PropertyCmd(Command):

299

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

300

def run_on_one_client(self, client, properties):

301

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

302

client.Set(client_interface, self.property, self.value_to_set,

303

dbus_interface=dbus.PROPERTIES_IFACE)

304

305

class ValueArgumentMixIn(object):

306

"""Mixin class for commands taking a value as argument"""

307

def __init__(self, value):

308

self.value_to_set = value

309

310

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

311

"""Mixin class for commands taking a value argument as

312

milliseconds."""

313

@property

314

def value_to_set(self):

315

return self._vts

316

@value_to_set.setter

317

def value_to_set(self, value):

318

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

319

self._vts = string_to_delta(value).total_seconds() * 1000

320

321

# Actual (non-abstract) command classes

322

323

class PrintTableCmd(PrintCmd):

324

def __init__(self, verbose=False):

325

self.verbose = verbose

326

327

def output(self, clients):

328

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

329

keywords = default_keywords

330

if self.verbose:

331

keywords = self.all_keywords

332

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

333

334

class TableOfClients(object):

335

tableheaders = {

336

"Name": "Name",

337

"Enabled": "Enabled",

338

"Timeout": "Timeout",

339

"LastCheckedOK": "Last Successful Check",

340

"LastApprovalRequest": "Last Approval Request",

341

"Created": "Created",

342

"Interval": "Interval",

343

"Host": "Host",

344

"Fingerprint": "Fingerprint",

345

"KeyID": "Key ID",

346

"CheckerRunning": "Check Is Running",

347

"LastEnabled": "Last Enabled",

348

"ApprovalPending": "Approval Is Pending",

349

"ApprovedByDefault": "Approved By Default",

350

"ApprovalDelay": "Approval Delay",

351

"ApprovalDuration": "Approval Duration",

352

"Checker": "Checker",

353

"ExtendedTimeout": "Extended Timeout",

354

"Expires": "Expires",

355

"LastCheckerStatus": "Last Checker Status",

356

}

357

358

def __init__(self, clients, keywords, tableheaders=None):

359

self.clients = clients

360

self.keywords = keywords

361

if tableheaders is not None:

362

self.tableheaders = tableheaders

363

364

def __str__(self):

365

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

366

367

if sys.version_info.major == 2:

368

__unicode__ = __str__

369

def __str__(self):

370

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

371

372

def rows(self):

373

format_string = self.row_formatting_string()

374

rows = [self.header_line(format_string)]

375

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

376

for client in self.clients)

377

return rows

378

379

def row_formatting_string(self):

380

"Format string used to format table rows"

381

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

382

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

383

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

384

for client in self.clients)),

385

key=key)

386

for key in self.keywords)

387

388

def string_from_client(self, client, key):

389

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

390

391

@staticmethod

392

def valuetostring(value, keyword):

393

if isinstance(value, dbus.Boolean):

394

return "Yes" if value else "No"

395

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

396

"ApprovalDuration", "ExtendedTimeout"):

397

return milliseconds_to_string(value)

398

return str(value)

399

400

def header_line(self, format_string):

401

return format_string.format(**self.tableheaders)

402

403

def client_line(self, client, format_string):

404

return format_string.format(

405

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

406

for key in self.keywords})

407

408

409

410

class DumpJSONCmd(PrintCmd):

411

def output(self, clients):

412

data = {client["Name"]:

413

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

414

for key in self.all_keywords}

415

for client in clients.values()}

416

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

417

@staticmethod

418

def dbus_boolean_to_bool(value):

419

if isinstance(value, dbus.Boolean):

420

value = bool(value)

421

return value

422

423

class IsEnabledCmd(Command):

424

def run_on_one_client(self, client, properties):

425

if self.is_enabled(client, properties):

426

sys.exit(0)

76

timevalue = datetime.timedelta(0)

77

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

78

79

for s in regexp.findall(interval):

80

try:

81

suffix = unicode(s[-1])

82

value = int(s[:-1])

83

if suffix == u"d":

84

delta = datetime.timedelta(value)

85

elif suffix == u"s":

86

delta = datetime.timedelta(0, value)

87

elif suffix == u"m":

88

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

89

elif suffix == u"h":

90

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

91

elif suffix == u"w":

92

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

93

else:

94

raise ValueError

95

except (ValueError, IndexError):

96

raise ValueError

97

timevalue += delta

98

return timevalue

99

100

def print_clients(clients):

101

def valuetostring(value, keyword):

102

if type(value) is dbus.Boolean:

103

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

104

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

105

return milliseconds_to_string(value)

106

return unicode(value)

107

108

# Create format string to print table rows

109

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

110

max(len(tablewords[key]),

111

max(len(valuetostring(client[key],

112

key))

113

for client in

114

clients))

115

for key in keywords)

116

# Print header line

117

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

118

for client in clients:

119

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

120

for key in keywords)

121

122

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

123

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

124

help="Print all fields")

125

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

126

help="Enable client")

127

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

128

help="disable client")

129

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

130

help="Bump timeout for client")

131

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

132

help="Start checker for client")

133

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

134

help="Stop checker for client")

135

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

136

help="Check if client is enabled")

137

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

138

help="Remove client")

139

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

140

help="Set checker command for client")

141

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

142

help="Set timeout for client")

143

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

144

help="Set checker interval for client")

145

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

146

help="Set host for client")

147

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

148

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

149

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

150

help="Approve any current client request")

151

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

152

help="Deny any current client request")

153

options, client_names = parser.parse_args()

154

155

# Compile list of clients to process

156

clients=[]

157

for name in client_names:

158

for path, client in mandos_clients.iteritems():

159

if client['name'] == name:

160

client_objc = bus.get_object(busname, path)

161

clients.append(client_objc)

162

break

163

else:

164

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

427

165

sys.exit(1)

428

def is_enabled(self, client, properties):

429

return bool(properties["Enabled"])

430

431

class RemoveCmd(Command):

432

def run_on_one_client(self, client, properties):

433

self.mandos.RemoveClient(client.__dbus_object_path__)

434

435

class ApproveCmd(Command):

436

def run_on_one_client(self, client, properties):

437

client.Approve(dbus.Boolean(True),

438

dbus_interface=client_interface)

439

440

class DenyCmd(Command):

441

def run_on_one_client(self, client, properties):

442

client.Approve(dbus.Boolean(False),

443

dbus_interface=client_interface)

444

445

class EnableCmd(PropertyCmd):

446

property = "Enabled"

447

value_to_set = dbus.Boolean(True)

448

449

class DisableCmd(PropertyCmd):

450

property = "Enabled"

451

value_to_set = dbus.Boolean(False)

452

453

class BumpTimeoutCmd(PropertyCmd):

454

property = "LastCheckedOK"

455

value_to_set = ""

456

457

class StartCheckerCmd(PropertyCmd):

458

property = "CheckerRunning"

459

value_to_set = dbus.Boolean(True)

460

461

class StopCheckerCmd(PropertyCmd):

462

property = "CheckerRunning"

463

value_to_set = dbus.Boolean(False)

464

465

class ApproveByDefaultCmd(PropertyCmd):

466

property = "ApprovedByDefault"

467

value_to_set = dbus.Boolean(True)

468

469

class DenyByDefaultCmd(PropertyCmd):

470

property = "ApprovedByDefault"

471

value_to_set = dbus.Boolean(False)

472

473

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

474

property = "Checker"

475

476

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

477

property = "Host"

478

479

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

480

@property

481

def value_to_set(self):

482

return self._vts

483

@value_to_set.setter

484

def value_to_set(self, value):

485

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

486

self._vts = value.read()

487

value.close()

488

property = "Secret"

489

490

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

491

property = "Timeout"

492

493

class SetExtendedTimeoutCmd(PropertyCmd,

494

MillisecondsValueArgumentMixIn):

495

property = "ExtendedTimeout"

496

497

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

498

property = "Interval"

499

500

class SetApprovalDelayCmd(PropertyCmd,

501

MillisecondsValueArgumentMixIn):

502

property = "ApprovalDelay"

503

504

class SetApprovalDurationCmd(PropertyCmd,

505

MillisecondsValueArgumentMixIn):

506

property = "ApprovalDuration"

507

508

def add_command_line_options(parser):

509

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

510

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

511

help="show version number and exit")

512

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

513

help="Select all clients")

514

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

515

help="Print all fields")

516

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

517

help="Dump client data in JSON format")

518

enable_disable = parser.add_mutually_exclusive_group()

519

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

520

help="Enable client")

521

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

522

action="store_true",

523

help="disable client")

524

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

525

help="Bump timeout for client")

526

start_stop_checker = parser.add_mutually_exclusive_group()

527

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

528

action="store_true",

529

help="Start checker for client")

530

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

531

action="store_true",

532

help="Stop checker for client")

533

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

534

help="Check if client is enabled")

535

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

536

help="Remove client")

537

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

538

help="Set checker command for client")

539

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

540

help="Set timeout for client")

541

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

542

help="Set extended timeout for client")

543

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

544

help="Set checker interval for client")

545

approve_deny_default = parser.add_mutually_exclusive_group()

546

approve_deny_default.add_argument(

547

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

548

default=None, dest="approved_by_default",

549

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

550

approve_deny_default.add_argument(

551

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

552

dest="approved_by_default",

553

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

554

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

555

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

556

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

557

help="Set duration of one client approval")

558

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

559

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

560

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

561

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

562

approve_deny = parser.add_mutually_exclusive_group()

563

approve_deny.add_argument(

564

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

565

help="Approve any current client request")

566

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

567

help="Deny any current client request")

568

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

569

help="Run self-test")

570

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

571

572

573

def commands_from_options(options):

574

575

commands = []

576

577

if options.dump_json:

578

commands.append(DumpJSONCmd())

579

166

167

if not clients and mandos_clients.values():

168

keywords = defaultkeywords

169

if options.all:

170

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

171

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

172

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

173

print_clients(mandos_clients.values())

174

175

# Process each client in the list by all selected options

176

for client in clients:

177

if options.remove:

178

mandos_serv.RemoveClient(client.__dbus_object_path__)

580

179

if options.enable:

581

commands.append(EnableCmd())

582

180

client.Enable(dbus_interface=client_interface)

583

181

if options.disable:

584

commands.append(DisableCmd())

585

182

client.Disable(dbus_interface=client_interface)

586

183

if options.bump_timeout:

587

commands.append(BumpTimeoutCmd())

588

184

client.CheckedOK(dbus_interface=client_interface)

589

185

if options.start_checker:

590

commands.append(StartCheckerCmd())

591

186

client.StartChecker(dbus_interface=client_interface)

592

187

if options.stop_checker:

593

commands.append(StopCheckerCmd())

594

188

client.StopChecker(dbus_interface=client_interface)

595

189

if options.is_enabled:

596

commands.append(IsEnabledCmd())

597

598

if options.remove:

599

commands.append(RemoveCmd())

600

601

if options.checker is not None:

602

commands.append(SetCheckerCmd(options.checker))

603

604

if options.timeout is not None:

605

commands.append(SetTimeoutCmd(options.timeout))

606

607

if options.extended_timeout:

608

commands.append(

609

SetExtendedTimeoutCmd(options.extended_timeout))

610

611

if options.interval is not None:

612

commands.append(SetIntervalCmd(options.interval))

613

614

if options.approved_by_default is not None:

615

if options.approved_by_default:

616

commands.append(ApproveByDefaultCmd())

617

else:

618

commands.append(DenyByDefaultCmd())

619

620

if options.approval_delay is not None:

621

commands.append(SetApprovalDelayCmd(options.approval_delay))

622

623

if options.approval_duration is not None:

624

commands.append(

625

SetApprovalDurationCmd(options.approval_duration))

626

627

if options.host is not None:

628

commands.append(SetHostCmd(options.host))

629

630

if options.secret is not None:

631

commands.append(SetSecretCmd(options.secret))

632

190

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

191

u"Enabled",

192

dbus_interface=dbus.PROPERTIES_IFACE)

193

else 1)

194

if options.checker:

195

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

196

dbus_interface=dbus.PROPERTIES_IFACE)

197

if options.host:

198

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

199

dbus_interface=dbus.PROPERTIES_IFACE)

200

if options.interval:

201

client.Set(client_interface, u"Interval",

202

timedelta_to_milliseconds

203

(string_to_delta(options.interval)),

204

dbus_interface=dbus.PROPERTIES_IFACE)

205

if options.timeout:

206

client.Set(client_interface, u"Timeout",

207

timedelta_to_milliseconds(string_to_delta

208

(options.timeout)),

209

dbus_interface=dbus.PROPERTIES_IFACE)

210

if options.secret:

211

client.Set(client_interface, u"Secret",

212

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

213

dbus_interface=dbus.PROPERTIES_IFACE)

633

214

if options.approve:

634

commands.append(ApproveCmd())

635

215

client.Approve(dbus.Boolean(True),

216

dbus_interface=client_interface)

636

217

if options.deny:

637

commands.append(DenyCmd())

638

639

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

640

# optionally verbosely

641

if not commands:

642

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

643

644

return commands

645

646

647

def main():

648

parser = argparse.ArgumentParser()

649

650

add_command_line_options(parser)

651

652

options = parser.parse_args()

653

654

def has_actions(options):

655

return any((options.enable,

656

options.disable,

657

options.bump_timeout,

658

options.start_checker,

659

options.stop_checker,

660

options.is_enabled,

661

options.remove,

662

options.checker is not None,

663

options.timeout is not None,

664

options.extended_timeout is not None,

665

options.interval is not None,

666

options.approved_by_default is not None,

667

options.approval_delay is not None,

668

options.approval_duration is not None,

669

options.host is not None,

670

options.secret is not None,

671

options.approve,

672

options.deny))

673

674

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

675

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

676

if options.verbose and has_actions(options):

677

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

678

if options.dump_json and (options.verbose

679

or has_actions(options)):

680

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

681

if options.all and not has_actions(options):

682

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

683

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

684

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

685

686

clientnames = options.client

687

688

try:

689

bus = dbus.SystemBus()

690

mandos_dbus_objc = bus.get_object(busname, server_path)

691

except dbus.exceptions.DBusException:

692

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

693

sys.exit(1)

694

695

mandos_serv = dbus.Interface(mandos_dbus_objc,

696

dbus_interface=server_interface)

697

mandos_serv_object_manager = dbus.Interface(

698

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

699

700

# Filter out log message from dbus module

701

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

702

class NullFilter(logging.Filter):

703

def filter(self, record):

704

return False

705

dbus_filter = NullFilter()

706

try:

707

dbus_logger.addFilter(dbus_filter)

708

mandos_clients = {path: ifs_and_props[client_interface]

709

for path, ifs_and_props in

710

mandos_serv_object_manager

711

.GetManagedObjects().items()

712

if client_interface in ifs_and_props}

713

except dbus.exceptions.DBusException as e:

714

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

715

"\n%s", e)

716

sys.exit(1)

717

finally:

718

# restore dbus logger

719

dbus_logger.removeFilter(dbus_filter)

720

721

# Compile dict of (clients: properties) to process

722

clients = {}

723

724

if not clientnames:

725

clients = {bus.get_object(busname, path): properties

726

for path, properties in mandos_clients.items()}

727

else:

728

for name in clientnames:

729

for path, client in mandos_clients.items():

730

if client["Name"] == name:

731

client_objc = bus.get_object(busname, path)

732

clients[client_objc] = client

733

break

734

else:

735

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

736

sys.exit(1)

737

738

# Run all commands on clients

739

commands = commands_from_options(options)

740

for command in commands:

741

command.run(mandos_serv, clients)

742

743

744

class Test_milliseconds_to_string(unittest.TestCase):

745

def test_all(self):

746

self.assertEqual(milliseconds_to_string(93785000),

747

"1T02:03:05")

748

def test_no_days(self):

749

self.assertEqual(milliseconds_to_string(7385000), "02:03:05")

750

def test_all_zero(self):

751

self.assertEqual(milliseconds_to_string(0), "00:00:00")

752

def test_no_fractional_seconds(self):

753

self.assertEqual(milliseconds_to_string(400), "00:00:00")

754

self.assertEqual(milliseconds_to_string(900), "00:00:00")

755

self.assertEqual(milliseconds_to_string(1900), "00:00:01")

756

757

class Test_string_to_delta(unittest.TestCase):

758

def test_handles_basic_rfc3339(self):

759

self.assertEqual(string_to_delta("PT0S"),

760

datetime.timedelta())

761

self.assertEqual(string_to_delta("P0D"),

762

datetime.timedelta())

763

self.assertEqual(string_to_delta("PT1S"),

764

datetime.timedelta(0, 1))

765

self.assertEqual(string_to_delta("PT2H"),

766

datetime.timedelta(0, 7200))

767

def test_falls_back_to_pre_1_6_1_with_warning(self):

768

# assertLogs only exists in Python 3.4

769

if hasattr(self, "assertLogs"):

770

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

771

value = string_to_delta("2h")

772

else:

773

class WarningFilter(logging.Filter):

774

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

775

def filter(self, record):

776

is_warning = record.levelno >= logging.WARNING

777

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

778

False)

779

return not is_warning

780

warning_filter = WarningFilter()

781

log.addFilter(warning_filter)

782

try:

783

value = string_to_delta("2h")

784

finally:

785

log.removeFilter(warning_filter)

786

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

787

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

788

789

790

class TestCmd(unittest.TestCase):

791

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

792

def setUp(self):

793

testcase = self

794

class MockClient(object):

795

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

796

self.__dbus_object_path__ = "objpath_{}".format(name)

797

self.attributes = attributes

798

self.attributes["Name"] = name

799

self.calls = []

800

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

801

testcase.assertEqual(interface, client_interface)

802

testcase.assertEqual(dbus_interface,

803

dbus.PROPERTIES_IFACE)

804

self.attributes[property] = value

805

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

806

testcase.assertEqual(interface, client_interface)

807

testcase.assertEqual(dbus_interface,

808

dbus.PROPERTIES_IFACE)

809

return self.attributes[property]

810

def Approve(self, approve, dbus_interface):

811

testcase.assertEqual(dbus_interface, client_interface)

812

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

813

dbus_interface)))

814

self.client = MockClient(

815

"foo",

816

KeyID=("92ed150794387c03ce684574b1139a65"

817

"94a34f895daaaf09fd8ea90a27cddb12"),

818

Secret=b"secret",

819

Host="foo.example.org",

820

Enabled=dbus.Boolean(True),

821

Timeout=300000,

822

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

823

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

824

Interval=120000,

825

Fingerprint=("778827225BA7DE539C5A"

826

"7CFA59CFF7CDBD9A5920"),

827

CheckerRunning=dbus.Boolean(False),

828

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

829

ApprovalPending=dbus.Boolean(False),

830

ApprovedByDefault=dbus.Boolean(True),

831

LastApprovalRequest="",

832

ApprovalDelay=0,

833

ApprovalDuration=1000,

834

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

835

ExtendedTimeout=900000,

836

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

837

LastCheckerStatus=0)

838

self.other_client = MockClient(

839

"barbar",

840

KeyID=("0558568eedd67d622f5c83b35a115f79"

841

"6ab612cff5ad227247e46c2b020f441c"),

842

Secret=b"secretbar",

843

Host="192.0.2.3",

844

Enabled=dbus.Boolean(True),

845

Timeout=300000,

846

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

847

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

848

Interval=120000,

849

Fingerprint=("3E393AEAEFB84C7E89E2"

850

"F547B3A107558FCA3A27"),

851

CheckerRunning=dbus.Boolean(True),

852

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

853

ApprovalPending=dbus.Boolean(False),

854

ApprovedByDefault=dbus.Boolean(False),

855

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

856

ApprovalDelay=30000,

857

ApprovalDuration=1000,

858

Checker=":",

859

ExtendedTimeout=900000,

860

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

861

LastCheckerStatus=-2)

862

self.clients = collections.OrderedDict(

863

[

864

(self.client, self.client.attributes),

865

(self.other_client, self.other_client.attributes),

866

])

867

self.one_client = {self.client: self.client.attributes}

868

869

class TestPrintTableCmd(TestCmd):

870

def test_normal(self):

871

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

872

expected_output = """

873

Name Enabled Timeout Last Successful Check

874

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

875

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

876

"""[1:-1]

877

self.assertEqual(output, expected_output)

878

def test_verbose(self):

879

output = PrintTableCmd(verbose=True).output(self.clients)

880

expected_output = """

881

Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status

882

foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0

883

barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2

884

"""[1:-1]

885

self.assertEqual(output, expected_output)

886

def test_one_client(self):

887

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

888

expected_output = """

889

Name Enabled Timeout Last Successful Check

890

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

891

"""[1:-1]

892

self.assertEqual(output, expected_output)

893

894

class TestDumpJSONCmd(TestCmd):

895

def setUp(self):

896

self.expected_json = {

897

"foo": {

898

"Name": "foo",

899

"KeyID": ("92ed150794387c03ce684574b1139a65"

900

"94a34f895daaaf09fd8ea90a27cddb12"),

901

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

902

"Enabled": True,

903

"Timeout": 300000,

904

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

905

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

906

"Interval": 120000,

907

"Fingerprint": ("778827225BA7DE539C5A"

908

"7CFA59CFF7CDBD9A5920"),

909

"CheckerRunning": False,

910

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

911

"ApprovalPending": False,

912

"ApprovedByDefault": True,

913

"LastApprovalRequest": "",

914

"ApprovalDelay": 0,

915

"ApprovalDuration": 1000,

916

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

917

"ExtendedTimeout": 900000,

918

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

919

"LastCheckerStatus": 0,

920

},

921

"barbar": {

922

"Name": "barbar",

923

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

924

"6ab612cff5ad227247e46c2b020f441c"),

925

"Host": "192.0.2.3",

926

"Enabled": True,

927

"Timeout": 300000,

928

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

929

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

930

"Interval": 120000,

931

"Fingerprint": ("3E393AEAEFB84C7E89E2"

932

"F547B3A107558FCA3A27"),

933

"CheckerRunning": True,

934

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

935

"ApprovalPending": False,

936

"ApprovedByDefault": False,

937

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

938

"ApprovalDelay": 30000,

939

"ApprovalDuration": 1000,

940

"Checker": ":",

941

"ExtendedTimeout": 900000,

942

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

943

"LastCheckerStatus": -2,

944

},

945

}

946

return super(TestDumpJSONCmd, self).setUp()

947

def test_normal(self):

948

json_data = json.loads(DumpJSONCmd().output(self.clients))

949

self.assertDictEqual(json_data, self.expected_json)

950

def test_one_client(self):

951

clients = self.one_client

952

json_data = json.loads(DumpJSONCmd().output(clients))

953

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

954

self.assertDictEqual(json_data, expected_json)

955

956

class TestIsEnabledCmd(TestCmd):

957

def test_is_enabled(self):

958

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

959

for client, properties in self.clients.items()))

960

def test_is_enabled_run_exits_successfully(self):

961

with self.assertRaises(SystemExit) as e:

962

IsEnabledCmd().run(None, self.one_client)

963

if e.exception.code is not None:

964

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

965

else:

966

self.assertIsNone(e.exception.code)

967

def test_is_enabled_run_exits_with_failure(self):

968

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

969

with self.assertRaises(SystemExit) as e:

970

IsEnabledCmd().run(None, self.one_client)

971

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

972

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

973

else:

974

self.assertIsNotNone(e.exception.code)

975

976

class TestRemoveCmd(TestCmd):

977

def test_remove(self):

978

class MockMandos(object):

979

def __init__(self):

980

self.calls = []

981

def RemoveClient(self, dbus_path):

982

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

983

mandos = MockMandos()

984

super(TestRemoveCmd, self).setUp()

985

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

986

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

987

for client in self.clients:

988

self.assertIn(("RemoveClient",

989

(client.__dbus_object_path__,)),

990

mandos.calls)

991

992

class TestApproveCmd(TestCmd):

993

def test_approve(self):

994

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

995

for client in self.clients:

996

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

997

client.calls)

998

999

class TestDenyCmd(TestCmd):

1000

def test_deny(self):

1001

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

1002

for client in self.clients:

1003

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

1004

client.calls)

1005

1006

class TestEnableCmd(TestCmd):

1007

def test_enable(self):

1008

for client in self.clients:

1009

client.attributes["Enabled"] = False

1010

1011

EnableCmd().run(None, self.clients)

1012

1013

for client in self.clients:

1014

self.assertTrue(client.attributes["Enabled"])

1015

1016

class TestDisableCmd(TestCmd):

1017

def test_disable(self):

1018

DisableCmd().run(None, self.clients)

1019

1020

for client in self.clients:

1021

self.assertFalse(client.attributes["Enabled"])

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

class TestPropertyCmd(TestCmd):

1028

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

1029

def runTest(self):

1030

if not hasattr(self, "command"):

1031

return

1032

values_to_get = getattr(self, "values_to_get",

1033

self.values_to_set)

1034

for value_to_set, value_to_get in zip(self.values_to_set,

1035

values_to_get):

1036

for client in self.clients:

1037

old_value = client.attributes[self.property]

1038

self.assertNotIsInstance(old_value, Unique)

1039

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

1040

self.run_command(value_to_set, self.clients)

1041

for client in self.clients:

1042

value = client.attributes[self.property]

1043

self.assertNotIsInstance(value, Unique)

1044

self.assertEqual(value, value_to_get)

1045

def run_command(self, value, clients):

1046

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

1047

1048

class TestBumpTimeoutCmd(TestPropertyCmd):

1049

command = BumpTimeoutCmd

1050

property = "LastCheckedOK"

1051

values_to_set = [""]

1052

1053

class TestStartCheckerCmd(TestPropertyCmd):

1054

command = StartCheckerCmd

1055

property = "CheckerRunning"

1056

values_to_set = [dbus.Boolean(True)]

1057

1058

class TestStopCheckerCmd(TestPropertyCmd):

1059

command = StopCheckerCmd

1060

property = "CheckerRunning"

1061

values_to_set = [dbus.Boolean(False)]

1062

1063

class TestApproveByDefaultCmd(TestPropertyCmd):

1064

command = ApproveByDefaultCmd

1065

property = "ApprovedByDefault"

1066

values_to_set = [dbus.Boolean(True)]

1067

1068

class TestDenyByDefaultCmd(TestPropertyCmd):

1069

command = DenyByDefaultCmd

1070

property = "ApprovedByDefault"

1071

values_to_set = [dbus.Boolean(False)]

1072

1073

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1074

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

1075

ValueArgumentMixIn"""

1076

def runTest(self):

1077

if type(self) is TestValueArgumentPropertyCmd:

1078

return

1079

return super(TestValueArgumentPropertyCmd, self).runTest()

1080

def run_command(self, value, clients):

1081

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

1082

1083

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1084

command = SetCheckerCmd

1085

property = "Checker"

1086

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

1087

1088

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1089

command = SetHostCmd

1090

property = "Host"

1091

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

1092

1093

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1094

command = SetSecretCmd

1095

property = "Secret"

1096

values_to_set = [open("/dev/null", "rb"),

1097

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

1098

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

1099

1100

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1101

command = SetTimeoutCmd

1102

property = "Timeout"

1103

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1104

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

1105

1106

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1107

command = SetExtendedTimeoutCmd

1108

property = "ExtendedTimeout"

1109

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1110

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

1111

1112

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1113

command = SetIntervalCmd

1114

property = "Interval"

1115

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1116

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

1117

1118

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1119

command = SetApprovalDelayCmd

1120

property = "ApprovalDelay"

1121

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1122

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

1123

1124

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1125

command = SetApprovalDurationCmd

1126

property = "ApprovalDuration"

1127

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1128

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

1129

1130

class Test_command_from_options(unittest.TestCase):

1131

def setUp(self):

1132

self.parser = argparse.ArgumentParser()

1133

add_command_line_options(self.parser)

1134

def assert_command_from_args(self, args, command_cls, **cmd_attrs):

1135

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

1136

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

1137

options = self.parser.parse_args(args)

1138

commands = commands_from_options(options)

1139

self.assertEqual(len(commands), 1)

1140

command = commands[0]

1141

self.assertIsInstance(command, command_cls)

1142

for key, value in cmd_attrs.items():

1143

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

1144

def test_print_table(self):

1145

self.assert_command_from_args([], PrintTableCmd,

1146

verbose=False)

1147

1148

def test_print_table_verbose(self):

1149

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

1150

verbose=True)

1151

1152

def test_enable(self):

1153

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

1154

1155

def test_disable(self):

1156

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

1157

DisableCmd)

1158

1159

def test_bump_timeout(self):

1160

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

1161

BumpTimeoutCmd)

1162

1163

def test_start_checker(self):

1164

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

1165

StartCheckerCmd)

1166

1167

def test_stop_checker(self):

1168

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

1169

StopCheckerCmd)

1170

1171

def test_remove(self):

1172

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

1173

RemoveCmd)

1174

1175

def test_checker(self):

1176

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

1177

SetCheckerCmd, value_to_set=":")

1178

1179

def test_checker_empty(self):

1180

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

1181

SetCheckerCmd, value_to_set="")

1182

1183

def test_timeout(self):

1184

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

1185

SetTimeoutCmd,

1186

value_to_set=300000)

1187

1188

def test_extended_timeout(self):

1189

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

1190

"foo"],

1191

SetExtendedTimeoutCmd,

1192

value_to_set=900000)

1193

1194

def test_interval(self):

1195

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

1196

SetIntervalCmd,

1197

value_to_set=120000)

1198

1199

def test_approve_by_default(self):

1200

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

1201

ApproveByDefaultCmd)

1202

1203

def test_deny_by_default(self):

1204

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

1205

DenyByDefaultCmd)

1206

1207

def test_approval_delay(self):

1208

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

1209

"foo"], SetApprovalDelayCmd,

1210

value_to_set=30000)

1211

1212

def test_approval_duration(self):

1213

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

1214

"foo"], SetApprovalDurationCmd,

1215

value_to_set=1000)

1216

1217

def test_host(self):

1218

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

1219

"foo"], SetHostCmd,

1220

value_to_set="foo.example.org")

1221

1222

def test_secret_devnull(self):

1223

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

1224

"foo"], SetSecretCmd,

1225

value_to_set=b"")

1226

1227

def test_secret_tempfile(self):

1228

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

1229

value = b"secret\0xyzzy\nbar"

1230

f.write(value)

1231

f.seek(0)

1232

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

1233

"foo"], SetSecretCmd,

1234

value_to_set=value)

1235

1236

def test_approve(self):

1237

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

1238

ApproveCmd)

1239

1240

def test_deny(self):

1241

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

1242

1243

def test_dump_json(self):

1244

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

1245

1246

def test_is_enabled(self):

1247

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

1248

IsEnabledCmd)

1249

1250

1251

1252

def should_only_run_tests():

1253

parser = argparse.ArgumentParser(add_help=False)

1254

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

1255

args, unknown_args = parser.parse_known_args()

1256

run_tests = args.check

1257

if run_tests:

1258

# Remove --check argument from sys.argv

1259

sys.argv[1:] = unknown_args

1260

return run_tests

1261

1262

# Add all tests from doctest strings

1263

def load_tests(loader, tests, none):

1264

import doctest

1265

tests.addTests(doctest.DocTestSuite())

1266

return tests

1267

1268

if __name__ == "__main__":

1269

if should_only_run_tests():

1270

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

1271

unittest.main()

1272

else:

1273

main()

218

client.Approve(dbus.Boolean(False),

219

dbus_interface=client_interface)