/mandos/release : revision 237.7.601

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

12

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

13

14

tablewords = {

15

'name': u'Name',

16

'enabled': u'Enabled',

17

'timeout': u'Timeout',

18

'last_checked_ok': u'Last Successful Check',

19

'created': u'Created',

20

'interval': u'Interval',

21

'host': u'Host',

22

'fingerprint': u'Fingerprint',

23

'checker_running': u'Check Is Running',

24

'last_enabled': u'Last Enabled',

25

'checker': u'Checker',

26

}

27

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

28

'checker')

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.9"

35

bus = dbus.SystemBus()

36

mandos_dbus_objc = bus.get_object(busname, server_path)

37

mandos_serv = dbus.Interface(mandos_dbus_objc,

38

dbus_interface = server_interface)

39

mandos_clients = mandos_serv.GetAllClientsWithProperties()

40

41

def datetime_to_milliseconds(dt):

42

"Return the 'timeout' attribute in milliseconds"

43

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

44

+ (dt.seconds * 1000)

45

+ (dt.microseconds // 1000))

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

64

65

locale.setlocale(locale.LC_ALL, "")

66

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

46

81

47

82

def milliseconds_to_string(ms):

48

83

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

49

return "%s%02d:%02d:%02d" % (("%dT" % td.days) if td.days else "", # days

50

td.seconds // 3600, # hours

51

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

52

(td.seconds % 60)) # seconds

84

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

85

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

86

hours=td.seconds // 3600,

87

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

88

seconds=td.seconds % 60))

89

90

91

def rfc3339_duration_to_delta(duration):

92

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

93

94

>>> rfc3339_duration_to_delta("P7D")

95

datetime.timedelta(7)

96

>>> rfc3339_duration_to_delta("PT60S")

97

datetime.timedelta(0, 60)

98

>>> rfc3339_duration_to_delta("PT60M")

99

datetime.timedelta(0, 3600)

100

>>> rfc3339_duration_to_delta("P60M")

101

datetime.timedelta(1680)

102

>>> rfc3339_duration_to_delta("PT24H")

103

datetime.timedelta(1)

104

>>> rfc3339_duration_to_delta("P1W")

105

datetime.timedelta(7)

106

>>> rfc3339_duration_to_delta("PT5M30S")

107

datetime.timedelta(0, 330)

108

>>> rfc3339_duration_to_delta("P1DT3M20S")

109

datetime.timedelta(1, 200)

110

>>> # Can not be empty:

111

>>> rfc3339_duration_to_delta("")

112

Traceback (most recent call last):

113

...

114

ValueError: Invalid RFC 3339 duration: u''

115

>>> # Must start with "P":

116

>>> rfc3339_duration_to_delta("1D")

117

Traceback (most recent call last):

118

...

119

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

120

>>> # Must use correct order

121

>>> rfc3339_duration_to_delta("PT1S2M")

122

Traceback (most recent call last):

123

...

124

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

125

>>> # Time needs time marker

126

>>> rfc3339_duration_to_delta("P1H2S")

127

Traceback (most recent call last):

128

...

129

ValueError: Invalid RFC 3339 duration: u'P1H2S'

130

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

131

>>> rfc3339_duration_to_delta("P1D2W")

132

Traceback (most recent call last):

133

...

134

ValueError: Invalid RFC 3339 duration: u'P1D2W'

135

>>> rfc3339_duration_to_delta("P2W2H")

136

Traceback (most recent call last):

137

...

138

ValueError: Invalid RFC 3339 duration: u'P2W2H'

139

"""

140

141

# Parsing an RFC 3339 duration with regular expressions is not

142

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

143

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

144

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

145

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

146

# avoid excessive use of external libraries.

147

148

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

149

Token = collections.namedtuple("Token", (

150

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

151

# a "group" containing digits

152

"value", # datetime.timedelta or None

153

"followers")) # Tokens valid after this token

154

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

155

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

156

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

157

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

158

datetime.timedelta(seconds=1),

159

frozenset((token_end, )))

160

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

161

datetime.timedelta(minutes=1),

162

frozenset((token_second, token_end)))

163

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

164

datetime.timedelta(hours=1),

165

frozenset((token_minute, token_end)))

166

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

167

None,

168

frozenset((token_hour, token_minute,

169

token_second)))

170

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

171

datetime.timedelta(days=1),

172

frozenset((token_time, token_end)))

173

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

174

datetime.timedelta(weeks=4),

175

frozenset((token_day, token_end)))

176

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

177

datetime.timedelta(weeks=52),

178

frozenset((token_month, token_end)))

179

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

180

datetime.timedelta(weeks=1),

181

frozenset((token_end, )))

182

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

183

frozenset((token_year, token_month,

184

token_day, token_time,

185

token_week)))

186

# Define starting values:

187

# Value so far

188

value = datetime.timedelta()

189

found_token = None

190

# Following valid tokens

191

followers = frozenset((token_duration, ))

192

# String left to parse

193

s = duration

194

# Loop until end token is found

195

while found_token is not token_end:

196

# Search for any currently valid tokens

197

for token in followers:

198

match = token.regexp.match(s)

199

if match is not None:

200

# Token found

201

if token.value is not None:

202

# Value found, parse digits

203

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

204

# Add to value so far

205

value += factor * token.value

206

# Strip token from string

207

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

208

# Go to found token

209

found_token = token

210

# Set valid next tokens

211

followers = found_token.followers

212

break

213

else:

214

# No currently valid tokens were found

215

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

216

.format(duration))

217

# End token found

218

return value

53

219

54

220

55

221

def string_to_delta(interval):

56

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

57

58

>>> string_to_delta('7d')

222

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

223

224

try:

225

return rfc3339_duration_to_delta(interval)

226

except ValueError as e:

227

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

228

' '.join(e.args))

229

return parse_pre_1_6_1_interval(interval)

230

231

232

def parse_pre_1_6_1_interval(interval):

233

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

234

and return a datetime.timedelta

235

236

>>> parse_pre_1_6_1_interval('7d')

59

237

datetime.timedelta(7)

60

>>> string_to_delta('60s')

238

>>> parse_pre_1_6_1_interval('60s')

61

239

datetime.timedelta(0, 60)

62

>>> string_to_delta('60m')

240

>>> parse_pre_1_6_1_interval('60m')

63

241

datetime.timedelta(0, 3600)

64

>>> string_to_delta('24h')

242

>>> parse_pre_1_6_1_interval('24h')

65

243

datetime.timedelta(1)

66

>>> string_to_delta(u'1w')

244

>>> parse_pre_1_6_1_interval('1w')

67

245

datetime.timedelta(7)

68

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

246

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

69

247

datetime.timedelta(0, 330)

248

>>> parse_pre_1_6_1_interval('')

249

datetime.timedelta(0)

250

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

251

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

252

datetime.timedelta(2, 480, 18000)

253

70

254

"""

71

timevalue = datetime.timedelta(0)

72

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

73

74

for s in regexp.findall(interval):

75

try:

76

suffix = unicode(s[-1])

77

value = int(s[:-1])

78

if suffix == u"d":

79

delta = datetime.timedelta(value)

80

elif suffix == u"s":

81

delta = datetime.timedelta(0, value)

82

elif suffix == u"m":

83

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

84

elif suffix == u"h":

85

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

86

elif suffix == u"w":

87

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

88

else:

89

raise ValueError

90

except (ValueError, IndexError):

91

raise ValueError

92

timevalue += delta

93

return timevalue

94

95

def print_clients(clients):

96

def valuetostring(value, keyword):

97

if type(value) is dbus.Boolean:

98

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

99

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

100

return milliseconds_to_string(value)

101

return unicode(value)

102

103

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

104

max(len(tablewords[key]),

105

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

106

for client in

107

clients))

108

for key in keywords)

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

114

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

115

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

116

help="Print all fields")

117

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

118

help="Enable client")

119

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

120

help="disable client")

121

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

122

help="Bump timeout for client")

123

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

124

help="Start checker for client")

125

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

126

help="Stop checker for client")

127

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

128

help="Check if client is still valid")

129

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

130

help="Remove client")

131

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

132

help="Set checker command for client")

133

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

134

help="Set timeout for client")

135

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

136

help="Set checker interval for client")

137

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

138

help="Set host for client")

139

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

140

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

141

options, client_names = parser.parse_args()

142

143

# Compile list of clients to process

144

clients=[]

145

for name in client_names:

146

for path, client in mandos_clients.iteritems():

147

if client['name'] == name:

148

client_objc = bus.get_object(busname, path)

149

clients.append(dbus.Interface(client_objc,

150

dbus_interface

151

= client_interface))

152

break

153

else:

154

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

255

256

value = datetime.timedelta(0)

257

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

258

259

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

260

if suffix == "d":

261

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

262

elif suffix == "s":

263

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

264

elif suffix == "m":

265

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

266

elif suffix == "h":

267

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

268

elif suffix == "w":

269

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

270

elif suffix == "":

271

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

272

return value

273

274

275

## Classes for commands.

276

277

# Abstract classes first

278

class Command(object):

279

"""Abstract class for commands"""

280

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

281

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

282

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

283

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

284

self.mandos = mandos

285

for clientpath, properties in clients.items():

286

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

287

dbus_busname, str(clientpath))

288

client = bus.get_object(dbus_busname, clientpath)

289

self.run_on_one_client(client, properties)

290

291

class PrintCmd(Command):

292

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

293

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

294

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

295

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

296

"ApprovalPending", "ApprovedByDefault",

297

"LastApprovalRequest", "ApprovalDelay",

298

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

299

"Expires", "LastCheckerStatus")

300

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

301

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

302

def output(self, clients):

303

raise NotImplementedError()

304

305

class PropertyCmd(Command):

306

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

307

def run_on_one_client(self, client, properties):

308

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

309

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

310

client.__dbus_object_path__,

311

dbus.PROPERTIES_IFACE, client_dbus_interface,

312

self.propname, self.value_to_set

313

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

314

else bool(self.value_to_set))

315

client.Set(client_dbus_interface, self.propname,

316

self.value_to_set,

317

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

321

322

class PropertyValueCmd(PropertyCmd):

323

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

324

def __init__(self, value):

325

self.value_to_set = value

326

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

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

329

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

330

@property

331

def value_to_set(self):

332

return self._vts

333

@value_to_set.setter

334

def value_to_set(self, value):

335

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

336

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

337

338

# Actual (non-abstract) command classes

339

340

class PrintTableCmd(PrintCmd):

341

def __init__(self, verbose=False):

342

self.verbose = verbose

343

344

def output(self, clients):

345

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

346

"LastCheckedOK")

347

keywords = default_keywords

348

if self.verbose:

349

keywords = self.all_keywords

350

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

351

352

class TableOfClients(object):

353

tableheaders = {

354

"Name": "Name",

355

"Enabled": "Enabled",

356

"Timeout": "Timeout",

357

"LastCheckedOK": "Last Successful Check",

358

"LastApprovalRequest": "Last Approval Request",

359

"Created": "Created",

360

"Interval": "Interval",

361

"Host": "Host",

362

"Fingerprint": "Fingerprint",

363

"KeyID": "Key ID",

364

"CheckerRunning": "Check Is Running",

365

"LastEnabled": "Last Enabled",

366

"ApprovalPending": "Approval Is Pending",

367

"ApprovedByDefault": "Approved By Default",

368

"ApprovalDelay": "Approval Delay",

369

"ApprovalDuration": "Approval Duration",

370

"Checker": "Checker",

371

"ExtendedTimeout": "Extended Timeout",

372

"Expires": "Expires",

373

"LastCheckerStatus": "Last Checker Status",

374

}

375

376

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

377

self.clients = clients

378

self.keywords = keywords

379

if tableheaders is not None:

380

self.tableheaders = tableheaders

381

382

def __str__(self):

383

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

384

385

if sys.version_info.major == 2:

386

__unicode__ = __str__

387

def __str__(self):

388

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

389

390

def rows(self):

391

format_string = self.row_formatting_string()

392

rows = [self.header_line(format_string)]

393

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

394

for client in self.clients)

395

return rows

396

397

def row_formatting_string(self):

398

"Format string used to format table rows"

399

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

400

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

401

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

402

for client in self.clients)),

403

key=key)

404

for key in self.keywords)

405

406

def string_from_client(self, client, key):

407

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

408

409

@staticmethod

410

def valuetostring(value, keyword):

411

if isinstance(value, dbus.Boolean):

412

return "Yes" if value else "No"

413

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

414

"ApprovalDuration", "ExtendedTimeout"):

415

return milliseconds_to_string(value)

416

return str(value)

417

418

def header_line(self, format_string):

419

return format_string.format(**self.tableheaders)

420

421

def client_line(self, client, format_string):

422

return format_string.format(

423

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

424

for key in self.keywords})

425

426

427

428

class DumpJSONCmd(PrintCmd):

429

def output(self, clients):

430

data = {client["Name"]:

431

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

432

for key in self.all_keywords}

433

for client in clients.values()}

434

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

435

@staticmethod

436

def dbus_boolean_to_bool(value):

437

if isinstance(value, dbus.Boolean):

438

value = bool(value)

439

return value

440

441

class IsEnabledCmd(Command):

442

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

443

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

444

if self.is_enabled(client, properties):

445

sys.exit(0)

155

446

sys.exit(1)

156

157

if not clients and mandos_clients.values():

158

keywords = defaultkeywords

159

if options.all:

160

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

161

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

162

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

163

print_clients(mandos_clients.values())

164

165

# Process each client in the list by all selected options

166

for client in clients:

167

if options.remove:

168

mandos_serv.RemoveClient(client.__dbus_object_path__)

447

def is_enabled(self, client, properties):

448

return properties["Enabled"]

449

450

class RemoveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

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

453

server_dbus_path, server_dbus_interface,

454

str(client.__dbus_object_path__))

455

self.mandos.RemoveClient(client.__dbus_object_path__)

456

457

class ApproveCmd(Command):

458

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_dbus_interface)

461

client.Approve(dbus.Boolean(True),

462

dbus_interface=client_dbus_interface)

463

464

class DenyCmd(Command):

465

def run_on_one_client(self, client, properties):

466

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

467

client.__dbus_object_path__, client_dbus_interface)

468

client.Approve(dbus.Boolean(False),

469

dbus_interface=client_dbus_interface)

470

471

class EnableCmd(PropertyCmd):

472

propname = "Enabled"

473

value_to_set = dbus.Boolean(True)

474

475

class DisableCmd(PropertyCmd):

476

propname = "Enabled"

477

value_to_set = dbus.Boolean(False)

478

479

class BumpTimeoutCmd(PropertyCmd):

480

propname = "LastCheckedOK"

481

value_to_set = ""

482

483

class StartCheckerCmd(PropertyCmd):

484

propname = "CheckerRunning"

485

value_to_set = dbus.Boolean(True)

486

487

class StopCheckerCmd(PropertyCmd):

488

propname = "CheckerRunning"

489

value_to_set = dbus.Boolean(False)

490

491

class ApproveByDefaultCmd(PropertyCmd):

492

propname = "ApprovedByDefault"

493

value_to_set = dbus.Boolean(True)

494

495

class DenyByDefaultCmd(PropertyCmd):

496

propname = "ApprovedByDefault"

497

value_to_set = dbus.Boolean(False)

498

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

507

@property

508

def value_to_set(self):

509

return self._vts

510

@value_to_set.setter

511

def value_to_set(self, value):

512

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

513

self._vts = value.read()

514

value.close()

515

516

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

517

propname = "Timeout"

518

519

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

520

propname = "ExtendedTimeout"

521

522

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

523

propname = "Interval"

524

525

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

526

propname = "ApprovalDelay"

527

528

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

529

propname = "ApprovalDuration"

530

531

def add_command_line_options(parser):

532

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

533

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

534

help="show version number and exit")

535

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

536

help="Select all clients")

537

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

538

help="Print all fields")

539

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

540

help="Dump client data in JSON format")

541

enable_disable = parser.add_mutually_exclusive_group()

542

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

543

help="Enable client")

544

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

545

action="store_true",

546

help="disable client")

547

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

548

help="Bump timeout for client")

549

start_stop_checker = parser.add_mutually_exclusive_group()

550

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

551

action="store_true",

552

help="Start checker for client")

553

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

554

action="store_true",

555

help="Stop checker for client")

556

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

557

help="Check if client is enabled")

558

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

559

help="Remove client")

560

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

561

help="Set checker command for client")

562

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

563

help="Set timeout for client")

564

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

565

help="Set extended timeout for client")

566

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

567

help="Set checker interval for client")

568

approve_deny_default = parser.add_mutually_exclusive_group()

569

approve_deny_default.add_argument(

570

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

571

default=None, dest="approved_by_default",

572

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

573

approve_deny_default.add_argument(

574

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

575

dest="approved_by_default",

576

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

577

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

578

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

579

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

580

help="Set duration of one client approval")

581

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

582

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

583

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

584

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

585

approve_deny = parser.add_mutually_exclusive_group()

586

approve_deny.add_argument(

587

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

588

help="Approve any current client request")

589

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

590

help="Deny any current client request")

591

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

592

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

593

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

594

help="Run self-test")

595

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

596

597

598

def commands_from_options(options):

599

600

commands = []

601

602

if options.dump_json:

603

commands.append(DumpJSONCmd())

604

169

605

if options.enable:

170

client.Enable()

606

commands.append(EnableCmd())

607

171

608

if options.disable:

172

client.Disable()

609

commands.append(DisableCmd())

610

173

611

if options.bump_timeout:

174

client.BumpTimeout()

612

commands.append(BumpTimeoutCmd())

613

175

614

if options.start_checker:

176

client.StartChecker()

615

commands.append(StartCheckerCmd())

616

177

617

if options.stop_checker:

178

client.StopChecker()

179

if options.is_valid:

180

sys.exit(0 if client.IsStillValid() else 1)

181

if options.checker:

182

client.SetChecker(options.checker)

183

if options.host:

184

client.SetHost(options.host)

185

if options.interval:

186

client.SetInterval(datetime_to_milliseconds

187

(string_to_delta(options.interval)))

188

if options.timeout:

189

client.SetTimeout(datetime_to_milliseconds

190

(string_to_delta(options.timeout)))

191

if options.secret:

192

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

193

618

commands.append(StopCheckerCmd())

619

620

if options.is_enabled:

621

commands.append(IsEnabledCmd())

622

623

if options.checker is not None:

624

commands.append(SetCheckerCmd(options.checker))

625

626

if options.timeout is not None:

627

commands.append(SetTimeoutCmd(options.timeout))

628

629

if options.extended_timeout:

630

commands.append(

631

SetExtendedTimeoutCmd(options.extended_timeout))

632

633

if options.interval is not None:

634

commands.append(SetIntervalCmd(options.interval))

635

636

if options.approved_by_default is not None:

637

if options.approved_by_default:

638

commands.append(ApproveByDefaultCmd())

639

else:

640

commands.append(DenyByDefaultCmd())

641

642

if options.approval_delay is not None:

643

commands.append(SetApprovalDelayCmd(options.approval_delay))

644

645

if options.approval_duration is not None:

646

commands.append(

647

SetApprovalDurationCmd(options.approval_duration))

648

649

if options.host is not None:

650

commands.append(SetHostCmd(options.host))

651

652

if options.secret is not None:

653

commands.append(SetSecretCmd(options.secret))

654

655

if options.approve:

656

commands.append(ApproveCmd())

657

658

if options.deny:

659

commands.append(DenyCmd())

660

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

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

665

# optionally verbosely

666

if not commands:

667

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

668

669

return commands

670

671

672

def check_option_syntax(parser, options):

673

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

674

argparse"""

675

676

def has_actions(options):

677

return any((options.enable,

678

options.disable,

679

options.bump_timeout,

680

options.start_checker,

681

options.stop_checker,

682

options.is_enabled,

683

options.remove,

684

options.checker is not None,

685

options.timeout is not None,

686

options.extended_timeout is not None,

687

options.interval is not None,

688

options.approved_by_default is not None,

689

options.approval_delay is not None,

690

options.approval_duration is not None,

691

options.host is not None,

692

options.secret is not None,

693

options.approve,

694

options.deny))

695

696

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

697

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

698

if options.verbose and has_actions(options):

699

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

700

if options.dump_json and (options.verbose

701

or has_actions(options)):

702

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

703

if options.all and not has_actions(options):

704

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

705

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

706

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

712

713

714

def main():

715

parser = argparse.ArgumentParser()

716

717

add_command_line_options(parser)

718

719

options = parser.parse_args()

720

721

check_option_syntax(parser, options)

722

723

clientnames = options.client

724

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

try:

729

bus = dbus.SystemBus()

730

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

731

dbus_busname, server_dbus_path)

732

mandos_dbus_objc = bus.get_object(dbus_busname,

733

server_dbus_path)

734

except dbus.exceptions.DBusException:

735

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

736

sys.exit(1)

737

738

mandos_serv = dbus.Interface(mandos_dbus_objc,

739

dbus_interface=server_dbus_interface)

740

mandos_serv_object_manager = dbus.Interface(

741

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

742

743

# Filter out log message from dbus module

744

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

745

class NullFilter(logging.Filter):

746

def filter(self, record):

747

return False

748

dbus_filter = NullFilter()

749

try:

750

dbus_logger.addFilter(dbus_filter)

751

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

752

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

753

mandos_clients = {path: ifs_and_props[client_dbus_interface]

754

for path, ifs_and_props in

755

mandos_serv_object_manager

756

.GetManagedObjects().items()

757

if client_dbus_interface in ifs_and_props}

758

except dbus.exceptions.DBusException as e:

759

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

760

"\n%s", e)

761

sys.exit(1)

762

finally:

763

# restore dbus logger

764

dbus_logger.removeFilter(dbus_filter)

765

766

# Compile dict of (clients: properties) to process

767

clients = {}

768

769

if not clientnames:

770

clients = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

772

else:

773

for name in clientnames:

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

777

break

778

else:

779

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

780

sys.exit(1)

781

782

# Run all commands on clients

783

commands = commands_from_options(options)

784

for command in commands:

785

command.run(clients, bus, mandos_serv)

786

787

788

class Test_milliseconds_to_string(unittest.TestCase):

789

def test_all(self):

790

self.assertEqual(milliseconds_to_string(93785000),

791

"1T02:03:05")

792

def test_no_days(self):

793

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

794

def test_all_zero(self):

795

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

796

def test_no_fractional_seconds(self):

797

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

798

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

799

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

800

801

class Test_string_to_delta(unittest.TestCase):

802

def test_handles_basic_rfc3339(self):

803

self.assertEqual(string_to_delta("PT0S"),

804

datetime.timedelta())

805

self.assertEqual(string_to_delta("P0D"),

806

datetime.timedelta())

807

self.assertEqual(string_to_delta("PT1S"),

808

datetime.timedelta(0, 1))

809

self.assertEqual(string_to_delta("PT2H"),

810

datetime.timedelta(0, 7200))

811

def test_falls_back_to_pre_1_6_1_with_warning(self):

812

# assertLogs only exists in Python 3.4

813

if hasattr(self, "assertLogs"):

814

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

815

value = string_to_delta("2h")

816

else:

817

class WarningFilter(logging.Filter):

818

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

819

def filter(self, record):

820

is_warning = record.levelno >= logging.WARNING

821

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

822

False)

823

return not is_warning

824

warning_filter = WarningFilter()

825

log.addFilter(warning_filter)

826

try:

827

value = string_to_delta("2h")

828

finally:

829

log.removeFilter(warning_filter)

830

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

831

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

832

833

834

class TestCmd(unittest.TestCase):

835

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

836

def setUp(self):

837

testcase = self

838

class MockClient(object):

839

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

840

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

841

self.attributes = attributes

842

self.attributes["Name"] = name

843

self.calls = []

844

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

845

testcase.assertEqual(interface, client_dbus_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[propname] = value

849

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

850

testcase.assertEqual(interface, client_dbus_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[propname]

854

def Approve(self, approve, dbus_interface):

855

testcase.assertEqual(dbus_interface,

856

client_dbus_interface)

857

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

858

dbus_interface)))

859

self.client = MockClient(

860

"foo",

861

KeyID=("92ed150794387c03ce684574b1139a65"

862

"94a34f895daaaf09fd8ea90a27cddb12"),

863

Secret=b"secret",

864

Host="foo.example.org",

865

Enabled=dbus.Boolean(True),

866

Timeout=300000,

867

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

868

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

869

Interval=120000,

870

Fingerprint=("778827225BA7DE539C5A"

871

"7CFA59CFF7CDBD9A5920"),

872

CheckerRunning=dbus.Boolean(False),

873

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

874

ApprovalPending=dbus.Boolean(False),

875

ApprovedByDefault=dbus.Boolean(True),

876

LastApprovalRequest="",

877

ApprovalDelay=0,

878

ApprovalDuration=1000,

879

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

880

ExtendedTimeout=900000,

881

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

882

LastCheckerStatus=0)

883

self.other_client = MockClient(

884

"barbar",

885

KeyID=("0558568eedd67d622f5c83b35a115f79"

886

"6ab612cff5ad227247e46c2b020f441c"),

887

Secret=b"secretbar",

888

Host="192.0.2.3",

889

Enabled=dbus.Boolean(True),

890

Timeout=300000,

891

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

892

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

893

Interval=120000,

894

Fingerprint=("3E393AEAEFB84C7E89E2"

895

"F547B3A107558FCA3A27"),

896

CheckerRunning=dbus.Boolean(True),

897

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

898

ApprovalPending=dbus.Boolean(False),

899

ApprovedByDefault=dbus.Boolean(False),

900

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

901

ApprovalDelay=30000,

902

ApprovalDuration=1000,

903

Checker=":",

904

ExtendedTimeout=900000,

905

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

906

LastCheckerStatus=-2)

907

self.clients = collections.OrderedDict(

908

[

909

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

910

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

911

])

912

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

913

@property

914

def bus(self):

915

class Bus(object):

916

@staticmethod

917

def get_object(client_bus_name, path):

918

self.assertEqual(client_bus_name, dbus_busname)

919

return {

920

"/clients/foo": self.client,

921

"/clients/barbar": self.other_client,

922

}[path]

923

return Bus()

924

925

class TestPrintTableCmd(TestCmd):

926

def test_normal(self):

927

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

928

expected_output = "\n".join((

929

"Name Enabled Timeout Last Successful Check",

930

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

931

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

932

))

933

self.assertEqual(output, expected_output)

934

def test_verbose(self):

935

output = PrintTableCmd(verbose=True).output(

936

self.clients.values())

937

expected_output = "\n".join((

938

# First line (headers)

939

"Name Enabled Timeout Last Successful Check Created "

940

" Interval Host Key ID "

941

" Fingerprint "

942

" Check Is Running Last Enabl"

943

"ed Approval Is Pending Approved By Default Last A"

944

"pproval Request Approval Delay Approval Duration Checker"

945

" Extended Timeout Expires Last "

946

"Checker Status",

947

# Second line (client "foo")

948

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

949

"T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684"

950

"574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7"

951

"DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03"

952

"T00:00:00 No Yes "

953

" 00:00:00 00:00:01 fping -"

954

"q -- %(host)s 00:15:00 2019-02-04T00:00:00 0 "

955

" ",

956

# Third line (client "barbar")

957

"barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03"

958

"T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c8"

959

"3b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB8"

960

"4C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04"

961

"T00:00:00 No No 2019-0"

962

"1-03T00:00:00 00:00:30 00:00:01 : "

963

" 00:15:00 2019-02-05T00:00:00 -2 "

964

" ",

965

))

966

self.assertEqual(output, expected_output)

967

def test_one_client(self):

968

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

969

expected_output = """

970

Name Enabled Timeout Last Successful Check

971

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

972

"""[1:-1]

973

self.assertEqual(output, expected_output)

974

975

class TestDumpJSONCmd(TestCmd):

976

def setUp(self):

977

self.expected_json = {

978

"foo": {

979

"Name": "foo",

980

"KeyID": ("92ed150794387c03ce684574b1139a65"

981

"94a34f895daaaf09fd8ea90a27cddb12"),

982

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

983

"Enabled": True,

984

"Timeout": 300000,

985

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

986

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

987

"Interval": 120000,

988

"Fingerprint": ("778827225BA7DE539C5A"

989

"7CFA59CFF7CDBD9A5920"),

990

"CheckerRunning": False,

991

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

992

"ApprovalPending": False,

993

"ApprovedByDefault": True,

994

"LastApprovalRequest": "",

995

"ApprovalDelay": 0,

996

"ApprovalDuration": 1000,

997

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

998

"ExtendedTimeout": 900000,

999

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

1000

"LastCheckerStatus": 0,

1001

},

1002

"barbar": {

1003

"Name": "barbar",

1004

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1005

"6ab612cff5ad227247e46c2b020f441c"),

1006

"Host": "192.0.2.3",

1007

"Enabled": True,

1008

"Timeout": 300000,

1009

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

1010

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

1011

"Interval": 120000,

1012

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1013

"F547B3A107558FCA3A27"),

1014

"CheckerRunning": True,

1015

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

1016

"ApprovalPending": False,

1017

"ApprovedByDefault": False,

1018

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

1019

"ApprovalDelay": 30000,

1020

"ApprovalDuration": 1000,

1021

"Checker": ":",

1022

"ExtendedTimeout": 900000,

1023

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

1024

"LastCheckerStatus": -2,

1025

},

1026

}

1027

return super(TestDumpJSONCmd, self).setUp()

1028

def test_normal(self):

1029

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

1030

self.assertDictEqual(json_data, self.expected_json)

1031

def test_one_client(self):

1032

clients = self.one_client

1033

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

1034

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

1035

self.assertDictEqual(json_data, expected_json)

1036

1037

class TestIsEnabledCmd(TestCmd):

1038

def test_is_enabled(self):

1039

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

1040

properties)

1041

for client, properties

1042

in self.clients.items()))

1043

def test_is_enabled_run_exits_successfully(self):

1044

with self.assertRaises(SystemExit) as e:

1045

IsEnabledCmd().run(self.one_client)

1046

if e.exception.code is not None:

1047

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

1048

else:

1049

self.assertIsNone(e.exception.code)

1050

def test_is_enabled_run_exits_with_failure(self):

1051

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

1052

with self.assertRaises(SystemExit) as e:

1053

IsEnabledCmd().run(self.one_client)

1054

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

1055

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

1056

else:

1057

self.assertIsNotNone(e.exception.code)

1058

1059

class TestRemoveCmd(TestCmd):

1060

def test_remove(self):

1061

class MockMandos(object):

1062

def __init__(self):

1063

self.calls = []

1064

def RemoveClient(self, dbus_path):

1065

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

1066

mandos = MockMandos()

1067

super(TestRemoveCmd, self).setUp()

1068

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

1069

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

1070

for clientpath in self.clients:

1071

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

1072

mandos.calls)

1073

1074

class TestApproveCmd(TestCmd):

1075

def test_approve(self):

1076

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

1077

for clientpath in self.clients:

1078

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

1079

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

1080

client.calls)

1081

1082

class TestDenyCmd(TestCmd):

1083

def test_deny(self):

1084

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

1085

for clientpath in self.clients:

1086

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

1087

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

1088

client.calls)

1089

1090

class TestEnableCmd(TestCmd):

1091

def test_enable(self):

1092

for clientpath in self.clients:

1093

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

1094

client.attributes["Enabled"] = False

1095

1096

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

1097

1098

for clientpath in self.clients:

1099

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

1100

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

1101

1102

class TestDisableCmd(TestCmd):

1103

def test_disable(self):

1104

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

1105

for clientpath in self.clients:

1106

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

1107

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

1108

1109

class Unique(object):

1110

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

1111

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

1112

1113

class TestPropertyCmd(TestCmd):

1114

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

1115

def runTest(self):

1116

if not hasattr(self, "command"):

1117

return

1118

values_to_get = getattr(self, "values_to_get",

1119

self.values_to_set)

1120

for value_to_set, value_to_get in zip(self.values_to_set,

1121

values_to_get):

1122

for clientpath in self.clients:

1123

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

1124

old_value = client.attributes[self.propname]

1125

self.assertNotIsInstance(old_value, Unique)

1126

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

1127

self.run_command(value_to_set, self.clients)

1128

for clientpath in self.clients:

1129

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

1130

value = client.attributes[self.propname]

1131

self.assertNotIsInstance(value, Unique)

1132

self.assertEqual(value, value_to_get)

1133

def run_command(self, value, clients):

1134

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

1135

1136

class TestBumpTimeoutCmd(TestPropertyCmd):

1137

command = BumpTimeoutCmd

1138

propname = "LastCheckedOK"

1139

values_to_set = [""]

1140

1141

class TestStartCheckerCmd(TestPropertyCmd):

1142

command = StartCheckerCmd

1143

propname = "CheckerRunning"

1144

values_to_set = [dbus.Boolean(True)]

1145

1146

class TestStopCheckerCmd(TestPropertyCmd):

1147

command = StopCheckerCmd

1148

propname = "CheckerRunning"

1149

values_to_set = [dbus.Boolean(False)]

1150

1151

class TestApproveByDefaultCmd(TestPropertyCmd):

1152

command = ApproveByDefaultCmd

1153

propname = "ApprovedByDefault"

1154

values_to_set = [dbus.Boolean(True)]

1155

1156

class TestDenyByDefaultCmd(TestPropertyCmd):

1157

command = DenyByDefaultCmd

1158

propname = "ApprovedByDefault"

1159

values_to_set = [dbus.Boolean(False)]

1160

1161

class TestPropertyValueCmd(TestPropertyCmd):

1162

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

1163

def runTest(self):

1164

if type(self) is TestPropertyValueCmd:

1165

return

1166

return super(TestPropertyValueCmd, self).runTest()

1167

def run_command(self, value, clients):

1168

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

1169

1170

class TestSetCheckerCmd(TestPropertyValueCmd):

1171

command = SetCheckerCmd

1172

propname = "Checker"

1173

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

1174

1175

class TestSetHostCmd(TestPropertyValueCmd):

1176

command = SetHostCmd

1177

propname = "Host"

1178

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

1179

1180

class TestSetSecretCmd(TestPropertyValueCmd):

1181

command = SetSecretCmd

1182

propname = "Secret"

1183

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

1184

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

1185

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

1186

1187

class TestSetTimeoutCmd(TestPropertyValueCmd):

1188

command = SetTimeoutCmd

1189

propname = "Timeout"

1190

values_to_set = [datetime.timedelta(),

1191

datetime.timedelta(minutes=5),

1192

datetime.timedelta(seconds=1),

1193

datetime.timedelta(weeks=1),

1194

datetime.timedelta(weeks=52)]

1195

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

1196

1197

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1198

command = SetExtendedTimeoutCmd

1199

propname = "ExtendedTimeout"

1200

values_to_set = [datetime.timedelta(),

1201

datetime.timedelta(minutes=5),

1202

datetime.timedelta(seconds=1),

1203

datetime.timedelta(weeks=1),

1204

datetime.timedelta(weeks=52)]

1205

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

1206

1207

class TestSetIntervalCmd(TestPropertyValueCmd):

1208

command = SetIntervalCmd

1209

propname = "Interval"

1210

values_to_set = [datetime.timedelta(),

1211

datetime.timedelta(minutes=5),

1212

datetime.timedelta(seconds=1),

1213

datetime.timedelta(weeks=1),

1214

datetime.timedelta(weeks=52)]

1215

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

1216

1217

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1218

command = SetApprovalDelayCmd

1219

propname = "ApprovalDelay"

1220

values_to_set = [datetime.timedelta(),

1221

datetime.timedelta(minutes=5),

1222

datetime.timedelta(seconds=1),

1223

datetime.timedelta(weeks=1),

1224

datetime.timedelta(weeks=52)]

1225

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

1226

1227

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1228

command = SetApprovalDurationCmd

1229

propname = "ApprovalDuration"

1230

values_to_set = [datetime.timedelta(),

1231

datetime.timedelta(minutes=5),

1232

datetime.timedelta(seconds=1),

1233

datetime.timedelta(weeks=1),

1234

datetime.timedelta(weeks=52)]

1235

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

1236

1237

class Test_command_from_options(unittest.TestCase):

1238

def setUp(self):

1239

self.parser = argparse.ArgumentParser()

1240

add_command_line_options(self.parser)

1241

def assert_command_from_args(self, args, command_cls,

1242

**cmd_attrs):

1243

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

1244

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

1245

options = self.parser.parse_args(args)

1246

check_option_syntax(self.parser, options)

1247

commands = commands_from_options(options)

1248

self.assertEqual(len(commands), 1)

1249

command = commands[0]

1250

self.assertIsInstance(command, command_cls)

1251

for key, value in cmd_attrs.items():

1252

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

1253

def test_print_table(self):

1254

self.assert_command_from_args([], PrintTableCmd,

1255

verbose=False)

1256

1257

def test_print_table_verbose(self):

1258

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

1259

verbose=True)

1260

1261

def test_print_table_verbose_short(self):

1262

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

1263

verbose=True)

1264

1265

def test_enable(self):

1266

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

1267

1268

def test_enable_short(self):

1269

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

1270

1271

def test_disable(self):

1272

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

1273

DisableCmd)

1274

1275

def test_disable_short(self):

1276

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

1277

1278

def test_bump_timeout(self):

1279

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

1280

BumpTimeoutCmd)

1281

1282

def test_bump_timeout_short(self):

1283

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

1284

1285

def test_start_checker(self):

1286

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

1287

StartCheckerCmd)

1288

1289

def test_stop_checker(self):

1290

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

1291

StopCheckerCmd)

1292

1293

def test_remove(self):

1294

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

1295

RemoveCmd)

1296

1297

def test_remove_short(self):

1298

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

1299

1300

def test_checker(self):

1301

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

1302

SetCheckerCmd, value_to_set=":")

1303

1304

def test_checker_empty(self):

1305

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

1306

SetCheckerCmd, value_to_set="")

1307

1308

def test_checker_short(self):

1309

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

1310

SetCheckerCmd, value_to_set=":")

1311

1312

def test_timeout(self):

1313

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

1314

SetTimeoutCmd,

1315

value_to_set=300000)

1316

1317

def test_timeout_short(self):

1318

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

1319

SetTimeoutCmd,

1320

value_to_set=300000)

1321

1322

def test_extended_timeout(self):

1323

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

1324

"foo"],

1325

SetExtendedTimeoutCmd,

1326

value_to_set=900000)

1327

1328

def test_interval(self):

1329

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

1330

SetIntervalCmd,

1331

value_to_set=120000)

1332

1333

def test_interval_short(self):

1334

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

1335

SetIntervalCmd,

1336

value_to_set=120000)

1337

1338

def test_approve_by_default(self):

1339

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

1340

ApproveByDefaultCmd)

1341

1342

def test_deny_by_default(self):

1343

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

1344

DenyByDefaultCmd)

1345

1346

def test_approval_delay(self):

1347

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

1348

"foo"], SetApprovalDelayCmd,

1349

value_to_set=30000)

1350

1351

def test_approval_duration(self):

1352

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

1353

"foo"], SetApprovalDurationCmd,

1354

value_to_set=1000)

1355

1356

def test_host(self):

1357

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

1358

"foo"], SetHostCmd,

1359

value_to_set="foo.example.org")

1360

1361

def test_host_short(self):

1362

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

1363

"foo"], SetHostCmd,

1364

value_to_set="foo.example.org")

1365

1366

def test_secret_devnull(self):

1367

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

1368

"foo"], SetSecretCmd,

1369

value_to_set=b"")

1370

1371

def test_secret_tempfile(self):

1372

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

1373

value = b"secret\0xyzzy\nbar"

1374

f.write(value)

1375

f.seek(0)

1376

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

1377

"foo"], SetSecretCmd,

1378

value_to_set=value)

1379

1380

def test_secret_devnull_short(self):

1381

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

1382

SetSecretCmd, value_to_set=b"")

1383

1384

def test_secret_tempfile_short(self):

1385

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

1386

value = b"secret\0xyzzy\nbar"

1387

f.write(value)

1388

f.seek(0)

1389

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

1390

SetSecretCmd,

1391

value_to_set=value)

1392

1393

def test_approve(self):

1394

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

1395

ApproveCmd)

1396

1397

def test_approve_short(self):

1398

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

1399

1400

def test_deny(self):

1401

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

1402

1403

def test_deny_short(self):

1404

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

1405

1406

def test_dump_json(self):

1407

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

1408

1409

def test_is_enabled(self):

1410

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

1411

IsEnabledCmd)

1412

1413

def test_is_enabled_short(self):

1414

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

1415

1416

def test_deny_before_remove(self):

1417

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

1418

"foo"])

1419

check_option_syntax(self.parser, options)

1420

commands = commands_from_options(options)

1421

self.assertEqual(len(commands), 2)

1422

self.assertIsInstance(commands[0], DenyCmd)

1423

self.assertIsInstance(commands[1], RemoveCmd)

1424

1425

def test_deny_before_remove_reversed(self):

1426

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

1427

"--all"])

1428

check_option_syntax(self.parser, options)

1429

commands = commands_from_options(options)

1430

self.assertEqual(len(commands), 2)

1431

self.assertIsInstance(commands[0], DenyCmd)

1432

self.assertIsInstance(commands[1], RemoveCmd)

1433

1434

1435

class Test_check_option_syntax(unittest.TestCase):

1436

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

1437

# check_option_syntax()

1438

actions = {

1439

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

1440

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

1441

# never used

1442

"enable": True,

1443

"disable": True,

1444

"bump_timeout": True,

1445

"start_checker": True,

1446

"stop_checker": True,

1447

"is_enabled": True,

1448

"remove": True,

1449

"checker": "x",

1450

"timeout": datetime.timedelta(),

1451

"extended_timeout": datetime.timedelta(),

1452

"interval": datetime.timedelta(),

1453

"approved_by_default": True,

1454

"approval_delay": datetime.timedelta(),

1455

"approval_duration": datetime.timedelta(),

1456

"host": "x",

1457

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

1458

"approve": True,

1459

"deny": True,

1460

}

1461

1462

def setUp(self):

1463

self.parser = argparse.ArgumentParser()

1464

add_command_line_options(self.parser)

1465

1466

@contextlib.contextmanager

1467

def assertParseError(self):

1468

with self.assertRaises(SystemExit) as e:

1469

with self.temporarily_suppress_stderr():

1470

yield

1471

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

1472

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

1473

# /argparse.html#exiting-methods

1474

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

1475

1476

@staticmethod

1477

@contextlib.contextmanager

1478

def temporarily_suppress_stderr():

1479

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

1480

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

1481

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

1482

os.close(null)

1483

try:

1484

yield

1485

finally:

1486

# restore stderr

1487

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

1488

os.close(stderrcopy)

1489

1490

def check_option_syntax(self, options):

1491

check_option_syntax(self.parser, options)

1492

1493

def test_actions_requires_client_or_all(self):

1494

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

1495

options = self.parser.parse_args()

1496

setattr(options, action, value)

1497

with self.assertParseError():

1498

self.check_option_syntax(options)

1499

1500

def test_actions_conflicts_with_verbose(self):

1501

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

1502

options = self.parser.parse_args()

1503

setattr(options, action, value)

1504

options.verbose = True

1505

with self.assertParseError():

1506

self.check_option_syntax(options)

1507

1508

def test_dump_json_conflicts_with_verbose(self):

1509

options = self.parser.parse_args()

1510

options.dump_json = True

1511

options.verbose = True

1512

with self.assertParseError():

1513

self.check_option_syntax(options)

1514

1515

def test_dump_json_conflicts_with_action(self):

1516

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

1517

options = self.parser.parse_args()

1518

setattr(options, action, value)

1519

options.dump_json = True

1520

with self.assertParseError():

1521

self.check_option_syntax(options)

1522

1523

def test_all_can_not_be_alone(self):

1524

options = self.parser.parse_args()

1525

options.all = True

1526

with self.assertParseError():

1527

self.check_option_syntax(options)

1528

1529

def test_all_is_ok_with_any_action(self):

1530

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

1531

options = self.parser.parse_args()

1532

setattr(options, action, value)

1533

options.all = True

1534

self.check_option_syntax(options)

1535

1536

def test_is_enabled_fails_without_client(self):

1537

options = self.parser.parse_args()

1538

options.is_enabled = True

1539

with self.assertParseError():

1540

self.check_option_syntax(options)

1541

1542

def test_is_enabled_works_with_one_client(self):

1543

options = self.parser.parse_args()

1544

options.is_enabled = True

1545

options.client = ["foo"]

1546

self.check_option_syntax(options)

1547

1548

def test_is_enabled_fails_with_two_clients(self):

1549

options = self.parser.parse_args()

1550

options.is_enabled = True

1551

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

1552

with self.assertParseError():

1553

self.check_option_syntax(options)

1554

1555

def test_remove_can_only_be_combined_with_action_deny(self):

1556

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

1557

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

1558

continue

1559

options = self.parser.parse_args()

1560

setattr(options, action, value)

1561

options.all = True

1562

options.remove = True

1563

with self.assertParseError():

1564

self.check_option_syntax(options)

1565

1566

1567

1568

def should_only_run_tests():

1569

parser = argparse.ArgumentParser(add_help=False)

1570

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

1571

args, unknown_args = parser.parse_known_args()

1572

run_tests = args.check

1573

if run_tests:

1574

# Remove --check argument from sys.argv

1575

sys.argv[1:] = unknown_args

1576

return run_tests

1577

1578

# Add all tests from doctest strings

1579

def load_tests(loader, tests, none):

1580

import doctest

1581

tests.addTests(doctest.DocTestSuite())

1582

return tests

1583

1584

if __name__ == "__main__":

1585

if should_only_run_tests():

1586

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

1587

unittest.main()

1588

else:

1589

main()