/mandos/trunk : revision 1092

1

#!/usr/bin/python3 -bb

2

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

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

#

45

import io

46

import tempfile

47

import contextlib

48

49

if sys.version_info.major == 2:

50

__metaclass__ = type

51

52

try:

53

import pydbus

54

import gi

55

dbus_python = None

56

except ImportError:

57

import dbus as dbus_python

58

pydbus = None

59

class gi:

60

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

61

class repository:

62

class GLib:

63

class Error(Exception):

64

pass

48

import abc

49

50

import dbus as dbus_python

65

51

66

52

# Show warnings by default

67

53

if not sys.warnoptions:

81

67

82

68

locale.setlocale(locale.LC_ALL, "")

83

69

84

version = "1.8.9"

70

version = "1.8.3"

85

71

86

72

87

73

def main():

96

82

if options.debug:

97

83

log.setLevel(logging.DEBUG)

98

84

99

if pydbus is not None:

100

bus = pydbus_adapter.CachingBus(pydbus)

101

else:

102

bus = dbus_python_adapter.CachingBus(dbus_python)

85

bus = dbus_python_adapter.CachingBus(dbus_python)

103

86

104

87

try:

105

88

all_clients = bus.get_clients_and_properties()

139

122

help="Select all clients")

140

123

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

141

124

help="Print all fields")

142

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

143

action="append_const", default=[],

144

const=command.DumpJSON(),

125

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

145

126

help="Dump client data in JSON format")

146

127

enable_disable = parser.add_mutually_exclusive_group()

147

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

148

action="append_const", default=[],

149

const=command.Enable(),

128

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

150

129

help="Enable client")

151

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

152

action="append_const", default=[],

153

const=command.Disable(),

130

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

131

action="store_true",

154

132

help="disable client")

155

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

156

action="append_const", default=[],

157

const=command.BumpTimeout(),

133

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

158

134

help="Bump timeout for client")

159

135

start_stop_checker = parser.add_mutually_exclusive_group()

160

136

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

161

dest="commands",

162

action="append_const", default=[],

163

const=command.StartChecker(),

137

action="store_true",

164

138

help="Start checker for client")

165

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

166

action="append_const", default=[],

167

const=command.StopChecker(),

139

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

140

action="store_true",

168

141

help="Stop checker for client")

169

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

170

action="append_const", default=[],

171

const=command.IsEnabled(),

142

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

172

143

help="Check if client is enabled")

173

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

174

action="append_const", default=[],

175

const=command.Remove(),

144

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

176

145

help="Remove client")

177

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

178

action="append", default=[],

179

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

146

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

180

147

help="Set checker command for client")

181

parser.add_argument(

182

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

183

default=[], metavar="TIME",

184

type=command.SetTimeout.argparse(string_to_delta),

185

help="Set timeout for client")

186

parser.add_argument(

187

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

188

default=[], metavar="TIME",

189

type=command.SetExtendedTimeout.argparse(string_to_delta),

190

help="Set extended timeout for client")

191

parser.add_argument(

192

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

193

default=[], metavar="TIME",

194

type=command.SetInterval.argparse(string_to_delta),

195

help="Set checker interval for client")

148

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

149

help="Set timeout for client")

150

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

151

help="Set extended timeout for client")

152

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

153

help="Set checker interval for client")

196

154

approve_deny_default = parser.add_mutually_exclusive_group()

197

155

approve_deny_default.add_argument(

198

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

199

action="append_const", default=[],

200

const=command.ApproveByDefault(),

156

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

157

default=None, dest="approved_by_default",

201

158

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

202

159

approve_deny_default.add_argument(

203

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

204

action="append_const", default=[],

205

const=command.DenyByDefault(),

160

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

161

dest="approved_by_default",

206

162

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

207

parser.add_argument(

208

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

209

default=[], metavar="TIME",

210

type=command.SetApprovalDelay.argparse(string_to_delta),

211

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

212

parser.add_argument(

213

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

214

default=[], metavar="TIME",

215

type=command.SetApprovalDuration.argparse(string_to_delta),

216

help="Set duration of one client approval")

217

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

218

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

219

type=command.SetHost,

220

help="Set host for client")

221

parser.add_argument(

222

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

223

default=[], metavar="FILENAME",

224

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

225

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

163

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

164

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

165

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

166

help="Set duration of one client approval")

167

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

168

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

169

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

170

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

226

171

approve_deny = parser.add_mutually_exclusive_group()

227

172

approve_deny.add_argument(

228

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

229

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

173

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

230

174

help="Approve any current client request")

231

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

232

action="append_const", default=[],

233

const=command.Deny(),

175

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

234

176

help="Deny any current client request")

235

177

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

236

178

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

253

195

def rfc3339_duration_to_delta(duration):

254

196

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

255

197

256

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

257

True

258

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

259

True

260

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

261

True

262

>>> # 60 months

263

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

264

True

265

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

266

True

267

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

268

True

269

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

270

True

271

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

272

True

198

>>> rfc3339_duration_to_delta("P7D")

199

datetime.timedelta(7)

200

>>> rfc3339_duration_to_delta("PT60S")

201

datetime.timedelta(0, 60)

202

>>> rfc3339_duration_to_delta("PT60M")

203

datetime.timedelta(0, 3600)

204

>>> rfc3339_duration_to_delta("P60M")

205

datetime.timedelta(1680)

206

>>> rfc3339_duration_to_delta("PT24H")

207

datetime.timedelta(1)

208

>>> rfc3339_duration_to_delta("P1W")

209

datetime.timedelta(7)

210

>>> rfc3339_duration_to_delta("PT5M30S")

211

datetime.timedelta(0, 330)

212

>>> rfc3339_duration_to_delta("P1DT3M20S")

213

datetime.timedelta(1, 200)

273

214

>>> # Can not be empty:

274

215

>>> rfc3339_duration_to_delta("")

275

216

Traceback (most recent call last):

385

326

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

386

327

and return a datetime.timedelta

387

328

388

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

389

True

390

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

391

True

392

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

393

True

394

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

395

True

396

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

397

True

398

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

399

True

400

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

401

True

329

>>> parse_pre_1_6_1_interval('7d')

330

datetime.timedelta(7)

331

>>> parse_pre_1_6_1_interval('60s')

332

datetime.timedelta(0, 60)

333

>>> parse_pre_1_6_1_interval('60m')

334

datetime.timedelta(0, 3600)

335

>>> parse_pre_1_6_1_interval('24h')

336

datetime.timedelta(1)

337

>>> parse_pre_1_6_1_interval('1w')

338

datetime.timedelta(7)

339

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

340

datetime.timedelta(0, 330)

341

>>> parse_pre_1_6_1_interval('')

342

datetime.timedelta(0)

402

343

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

403

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

404

True

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

405

346

406

347

"""

407

348

428

369

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

429

370

argparse"""

430

371

431

def has_commands(options, commands=None):

432

if commands is None:

433

commands = (command.Enable,

434

command.Disable,

435

command.BumpTimeout,

436

command.StartChecker,

437

command.StopChecker,

438

command.IsEnabled,

439

command.Remove,

440

command.SetChecker,

441

command.SetTimeout,

442

command.SetExtendedTimeout,

443

command.SetInterval,

444

command.ApproveByDefault,

445

command.DenyByDefault,

446

command.SetApprovalDelay,

447

command.SetApprovalDuration,

448

command.SetHost,

449

command.SetSecret,

450

command.Approve,

451

command.Deny)

452

return any(isinstance(cmd, commands)

453

for cmd in options.commands)

372

def has_actions(options):

373

return any((options.enable,

374

options.disable,

375

options.bump_timeout,

376

options.start_checker,

377

options.stop_checker,

378

options.is_enabled,

379

options.remove,

380

options.checker is not None,

381

options.timeout is not None,

382

options.extended_timeout is not None,

383

options.interval is not None,

384

options.approved_by_default is not None,

385

options.approval_delay is not None,

386

options.approval_duration is not None,

387

options.host is not None,

388

options.secret is not None,

389

options.approve,

390

options.deny))

454

391

455

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

392

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

456

393

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

457

if options.verbose and has_commands(options):

394

if options.verbose and has_actions(options):

458

395

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

459

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

460

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

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

461

398

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

462

if options.all and not has_commands(options):

399

if options.all and not has_actions(options):

463

400

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

464

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

465

and len(options.client) > 1):

401

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

466

402

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

467

if (len(options.commands) > 1

468

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

469

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

470

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

471

472

473

class dbus:

474

475

class SystemBus:

403

if options.remove:

404

options.remove = False

405

if has_actions(options) and not options.deny:

406

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

407

options.remove = True

408

409

410

411

class dbus(object):

412

413

class SystemBus(object):

476

414

477

415

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

478

416

def get_managed_objects(self, busname, objectpath):

524

462

pass

525

463

526

464

527

class dbus_python_adapter:

465

class dbus_python_adapter(object):

528

466

529

467

class SystemBus(dbus.MandosBus):

530

468

"""Use dbus-python"""

575

513

for key, subval in value.items()}

576

514

return value

577

515

578

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

579

if key == "Secret":

580

if not isinstance(value, bytes):

581

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

582

value = self.dbus_python.ByteArray(value)

583

return self.set_property(self.busname, objectpath,

584

self.client_interface, key,

585

value)

586

516

587

class SilenceLogger:

517

class SilenceLogger(object):

588

518

"Simple context manager to silence a particular logger"

589

519

def __init__(self, loggername):

590

520

self.logger = logging.getLogger(loggername)

619

549

return new_object

620

550

621

551

622

class pydbus_adapter:

623

class SystemBus(dbus.MandosBus):

624

def __init__(self, module=pydbus):

625

self.pydbus = module

626

self.bus = self.pydbus.SystemBus()

627

628

@contextlib.contextmanager

629

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

630

try:

631

yield

632

except gi.repository.GLib.Error as e:

633

# This does what "raise from" would do

634

exc = exception_class(*e.args)

635

exc.__cause__ = e

636

raise exc

637

638

def call_method(self, methodname, busname, objectpath,

639

interface, *args):

640

proxy_object = self.get(busname, objectpath)

641

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

642

interface, methodname,

643

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

644

method = getattr(proxy_object[interface], methodname)

645

with self.convert_exception():

646

return method(*args)

647

648

def get(self, busname, objectpath):

649

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

650

busname, objectpath)

651

with self.convert_exception(dbus.ConnectFailed):

652

if sys.version_info.major <= 2:

653

with warnings.catch_warnings():

654

warnings.filterwarnings(

655

"ignore", "", DeprecationWarning,

656

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

657

return self.bus.get(busname, objectpath)

658

else:

659

return self.bus.get(busname, objectpath)

660

661

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

662

value):

663

proxy_object = self.get(busname, objectpath)

664

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

665

objectpath, self.properties_iface, interface,

666

key, value)

667

setattr(proxy_object[interface], key, value)

668

669

class CachingBus(SystemBus):

670

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

671

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

672

self.object_cache = {}

673

super(pydbus_adapter.CachingBus,

674

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

675

def get(self, busname, objectpath):

676

try:

677

return self.object_cache[(busname, objectpath)]

678

except KeyError:

679

new_object = (super(pydbus_adapter.CachingBus, self)

680

.get(busname, objectpath))

681

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

682

return new_object

683

684

685

552

def commands_from_options(options):

686

553

687

commands = list(options.commands)

688

689

def find_cmd(cmd, commands):

690

i = 0

691

for i, c in enumerate(commands):

692

if isinstance(c, cmd):

693

return i

694

return i+1

695

696

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

697

# to occur ahead of command.Remove.

698

index_of_remove = find_cmd(command.Remove, commands)

699

before_remove = commands[:index_of_remove]

700

after_remove = commands[index_of_remove:]

701

cleaned_after = []

702

for cmd in after_remove:

703

if isinstance(cmd, command.Deny):

704

before_remove.append(cmd)

554

commands = []

555

556

if options.is_enabled:

557

commands.append(command.IsEnabled())

558

559

if options.approve:

560

commands.append(command.Approve())

561

562

if options.deny:

563

commands.append(command.Deny())

564

565

if options.remove:

566

commands.append(command.Remove())

567

568

if options.dump_json:

569

commands.append(command.DumpJSON())

570

571

if options.enable:

572

commands.append(command.Enable())

573

574

if options.disable:

575

commands.append(command.Disable())

576

577

if options.bump_timeout:

578

commands.append(command.BumpTimeout())

579

580

if options.start_checker:

581

commands.append(command.StartChecker())

582

583

if options.stop_checker:

584

commands.append(command.StopChecker())

585

586

if options.approved_by_default is not None:

587

if options.approved_by_default:

588

commands.append(command.ApproveByDefault())

705

589

else:

706

cleaned_after.append(cmd)

707

if cleaned_after != after_remove:

708

commands = before_remove + cleaned_after

590

commands.append(command.DenyByDefault())

591

592

if options.checker is not None:

593

commands.append(command.SetChecker(options.checker))

594

595

if options.host is not None:

596

commands.append(command.SetHost(options.host))

597

598

if options.secret is not None:

599

commands.append(command.SetSecret(options.secret))

600

601

if options.timeout is not None:

602

commands.append(command.SetTimeout(options.timeout))

603

604

if options.extended_timeout:

605

commands.append(

606

command.SetExtendedTimeout(options.extended_timeout))

607

608

if options.interval is not None:

609

commands.append(command.SetInterval(options.interval))

610

611

if options.approval_delay is not None:

612

commands.append(

613

command.SetApprovalDelay(options.approval_delay))

614

615

if options.approval_duration is not None:

616

commands.append(

617

command.SetApprovalDuration(options.approval_duration))

709

618

710

619

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

711

620

# optionally verbosely

715

624

return commands

716

625

717

626

718

class command:

627

class command(object):

719

628

"""A namespace for command classes"""

720

629

721

class Base:

630

class Base(object):

722

631

"""Abstract base class for commands"""

723

632

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

724

633

"""Normal commands should implement run_on_one_client(),

787

696

keywords = self.all_keywords

788

697

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

789

698

790

class TableOfClients:

699

class TableOfClients(object):

791

700

tableheaders = {

792

701

"Name": "Name",

793

702

"Enabled": "Enabled",

926

835

def __init__(self, value):

927

836

self.value_to_set = value

928

837

929

@classmethod

930

def argparse(cls, argtype):

931

def cmdtype(arg):

932

return cls(argtype(arg))

933

return cmdtype

934

838

935

839

class SetChecker(PropertySetterValue):

936

840

propname = "Checker"

1024

928

"output"))

1025

929

1026

930

1027

class Unique:

931

class Unique(object):

1028

932

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

1029

933

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

1030

934

1062

966

1063

967

def test_actions_requires_client_or_all(self):

1064

968

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

1065

args = self.actionargs(action, value)

969

options = self.parser.parse_args()

970

setattr(options, action, value)

1066

971

with self.assertParseError():

1067

self.parse_args(args)

972

self.check_option_syntax(options)

1068

973

1069

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

974

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

1070

975

# check_option_syntax()

1071

976

actions = {

1072

"--enable": None,

1073

"--disable": None,

1074

"--bump-timeout": None,

1075

"--start-checker": None,

1076

"--stop-checker": None,

1077

"--is-enabled": None,

1078

"--remove": None,

1079

"--checker": "x",

1080

"--timeout": "PT0S",

1081

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

1082

"--interval": "PT0S",

1083

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

1084

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

1085

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

1086

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

1087

"--host": "hostname",

1088

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

1089

"--approve": None,

1090

"--deny": None,

977

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

978

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

979

# never used

980

"enable": True,

981

"disable": True,

982

"bump_timeout": True,

983

"start_checker": True,

984

"stop_checker": True,

985

"is_enabled": True,

986

"remove": True,

987

"checker": "x",

988

"timeout": datetime.timedelta(),

989

"extended_timeout": datetime.timedelta(),

990

"interval": datetime.timedelta(),

991

"approved_by_default": True,

992

"approval_delay": datetime.timedelta(),

993

"approval_duration": datetime.timedelta(),

994

"host": "x",

995

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

996

"approve": True,

997

"deny": True,

1091

998

}

1092

999

1093

@staticmethod

1094

def actionargs(action, value, *args):

1095

if value is not None:

1096

return [action, value] + list(args)

1097

else:

1098

return [action] + list(args)

1099

1100

1000

@contextlib.contextmanager

1101

1001

def assertParseError(self):

1102

1002

with self.assertRaises(SystemExit) as e:

1107

1007

# /argparse.html#exiting-methods

1108

1008

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

1109

1009

1110

def parse_args(self, args):

1111

options = self.parser.parse_args(args)

1112

check_option_syntax(self.parser, options)

1113

1114

1010

@staticmethod

1115

1011

@contextlib.contextmanager

1116

1012

def redirect_stderr_to_devnull():

1127

1023

1128

1024

def test_actions_all_conflicts_with_verbose(self):

1129

1025

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

1130

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

1131

"--verbose")

1026

options = self.parser.parse_args()

1027

setattr(options, action, value)

1028

options.all = True

1029

options.verbose = True

1132

1030

with self.assertParseError():

1133

self.parse_args(args)

1031

self.check_option_syntax(options)

1134

1032

1135

1033

def test_actions_with_client_conflicts_with_verbose(self):

1136

1034

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

1137

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

1138

"client")

1035

options = self.parser.parse_args()

1036

setattr(options, action, value)

1037

options.verbose = True

1038

options.client = ["client"]

1139

1039

with self.assertParseError():

1140

self.parse_args(args)

1040

self.check_option_syntax(options)

1141

1041

1142

1042

def test_dump_json_conflicts_with_verbose(self):

1143

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

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1144

1046

with self.assertParseError():

1145

self.parse_args(args)

1047

self.check_option_syntax(options)

1146

1048

1147

1049

def test_dump_json_conflicts_with_action(self):

1148

1050

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

1149

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

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1150

1054

with self.assertParseError():

1151

self.parse_args(args)

1055

self.check_option_syntax(options)

1152

1056

1153

1057

def test_all_can_not_be_alone(self):

1154

args = ["--all"]

1058

options = self.parser.parse_args()

1059

options.all = True

1155

1060

with self.assertParseError():

1156

self.parse_args(args)

1061

self.check_option_syntax(options)

1157

1062

1158

1063

def test_all_is_ok_with_any_action(self):

1159

1064

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

1160

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

1161

self.parse_args(args)

1065

options = self.parser.parse_args()

1066

setattr(options, action, value)

1067

options.all = True

1068

self.check_option_syntax(options)

1162

1069

1163

1070

def test_any_action_is_ok_with_one_client(self):

1164

1071

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

1165

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

1166

self.parse_args(args)

1072

options = self.parser.parse_args()

1073

setattr(options, action, value)

1074

options.client = ["client"]

1075

self.check_option_syntax(options)

1167

1076

1168

1077

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1169

1079

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

1170

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

1080

if action == "is_enabled":

1171

1081

continue

1172

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

1173

self.parse_args(args)

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1174

1085

1175

1086

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1176

1088

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

1177

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

1089

if action == "is_enabled":

1178

1090

continue

1179

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

1180

"client2")

1181

self.parse_args(args)

1091

setattr(options, action, value)

1092

options.client = ["client1", "client2"]

1093

self.check_option_syntax(options)

1182

1094

1183

1095

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1184

1096

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

1185

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

1097

if action == "is_enabled":

1186

1098

continue

1187

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

1188

"client2")

1189

self.parse_args(args)

1099

options = self.parser.parse_args()

1100

setattr(options, action, value)

1101

options.client = ["client1", "client2"]

1102

self.check_option_syntax(options)

1190

1103

1191

1104

def test_is_enabled_fails_without_client(self):

1192

args = ["--is-enabled"]

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1193

1107

with self.assertParseError():

1194

self.parse_args(args)

1108

self.check_option_syntax(options)

1195

1109

1196

1110

def test_is_enabled_fails_with_two_clients(self):

1197

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

1111

options = self.parser.parse_args()

1112

options.is_enabled = True

1113

options.client = ["client1", "client2"]

1198

1114

with self.assertParseError():

1199

self.parse_args(args)

1115

self.check_option_syntax(options)

1200

1116

1201

1117

def test_remove_can_only_be_combined_with_action_deny(self):

1202

1118

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

1203

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

1119

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

1204

1120

continue

1205

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

1206

"--remove")

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1207

1125

with self.assertParseError():

1208

self.parse_args(args)

1126

self.check_option_syntax(options)

1209

1127

1210

1128

1211

1129

class Test_dbus_exceptions(unittest.TestCase):

1314

1232

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1315

1233

1316

1234

def MockDBusPython_func(self, func):

1317

class mock_dbus_python:

1235

class mock_dbus_python(object):

1318

1236

"""mock dbus-python module"""

1319

class exceptions:

1237

class exceptions(object):

1320

1238

"""Pseudo-namespace"""

1321

1239

class DBusException(Exception):

1322

1240

pass

1323

class SystemBus:

1241

class SystemBus(object):

1324

1242

@staticmethod

1325

1243

def get_object(busname, objectpath):

1326

1244

DBusObject = collections.namedtuple(

1327

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

1245

"DBusObject", ("methodname",))

1328

1246

def method(*args, **kwargs):

1329

1247

self.assertEqual({"dbus_interface":

1330

1248

"interface"},

1331

1249

kwargs)

1332

1250

return func(*args)

1333

def set_property(interface, key, value,

1334

dbus_interface=None):

1335

self.assertEqual(

1336

"org.freedesktop.DBus.Properties",

1337

dbus_interface)

1338

self.assertEqual("Secret", key)

1339

return func(interface, key, value,

1340

dbus_interface=dbus_interface)

1341

return DBusObject(methodname=method,

1342

Set=set_property)

1343

class Boolean:

1251

return DBusObject(methodname=method)

1252

class Boolean(object):

1344

1253

def __init__(self, value):

1345

1254

self.value = bool(value)

1346

1255

def __bool__(self):

1351

1260

pass

1352

1261

class Dictionary(dict):

1353

1262

pass

1354

class ByteArray(bytes):

1355

pass

1356

1263

return mock_dbus_python

1357

1264

1358

1265

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

1535

1442

# Make sure the dbus logger was suppressed

1536

1443

self.assertEqual(0, counting_handler.count)

1537

1444

1538

def test_Set_Secret_sends_bytearray(self):

1539

ret = [None]

1540

def func(*args, **kwargs):

1541

ret[0] = (args, kwargs)

1542

mock_dbus_python = self.MockDBusPython_func(func)

1543

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1544

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

1545

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

1546

mock_dbus_python.ByteArray(b"value")),

1547

{"dbus_interface":

1548

"org.freedesktop.DBus.Properties"})

1549

self.assertEqual(expected_call, ret[0])

1550

if sys.version_info.major == 2:

1551

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

1552

mock_dbus_python.ByteArray)

1553

1554

1445

def test_get_object_converts_to_correct_exception(self):

1555

1446

bus = dbus_python_adapter.SystemBus(

1556

1447

self.fake_dbus_python_raises_exception_on_connect)

1558

1449

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

1559

1450

"objectpath", "interface")

1560

1451

1561

class fake_dbus_python_raises_exception_on_connect:

1452

class fake_dbus_python_raises_exception_on_connect(object):

1562

1453

"""fake dbus-python module"""

1563

class exceptions:

1454

class exceptions(object):

1564

1455

"""Pseudo-namespace"""

1565

1456

class DBusException(Exception):

1566

1457

pass

1574

1465

1575

1466

1576

1467

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1577

class mock_dbus_python:

1468

class mock_dbus_python(object):

1578

1469

"""mock dbus-python modules"""

1579

class SystemBus:

1470

class SystemBus(object):

1580

1471

@staticmethod

1581

1472

def get_object(busname, objectpath):

1582

1473

return Unique()

1625

1516

self.assertIs(obj1, obj1b)

1626

1517

1627

1518

1628

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1629

1630

def Stub_pydbus_func(self, func):

1631

class stub_pydbus:

1632

"""stub pydbus module"""

1633

class SystemBus:

1634

@staticmethod

1635

def get(busname, objectpath):

1636

DBusObject = collections.namedtuple(

1637

"DBusObject", ("methodname",))

1638

return {"interface":

1639

DBusObject(methodname=func)}

1640

return stub_pydbus

1641

1642

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

1643

interface, *args):

1644

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

1645

return bus.call_method(methodname, busname, objectpath,

1646

interface, *args)

1647

1648

def test_call_method_returns(self):

1649

expected_method_return = Unique()

1650

method_args = (Unique(), Unique())

1651

def func(*args):

1652

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

1653

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

1654

self.assertIs(marg, arg)

1655

return expected_method_return

1656

stub_pydbus = self.Stub_pydbus_func(func)

1657

bus = pydbus_adapter.SystemBus(stub_pydbus)

1658

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

1659

"objectpath", "interface",

1660

*method_args)

1661

self.assertIs(ret, expected_method_return)

1662

1663

def test_call_method_handles_exception(self):

1664

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

1665

1666

def func():

1667

raise gi.repository.GLib.Error()

1668

1669

stub_pydbus = self.Stub_pydbus_func(func)

1670

bus = pydbus_adapter.SystemBus(stub_pydbus)

1671

1672

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

1673

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

1674

"objectpath", "interface")

1675

1676

self.assertNotIsInstance(e, dbus.ConnectFailed)

1677

1678

def test_get_converts_to_correct_exception(self):

1679

bus = pydbus_adapter.SystemBus(

1680

self.fake_pydbus_raises_exception_on_connect)

1681

with self.assertRaises(dbus.ConnectFailed):

1682

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

1683

"objectpath", "interface")

1684

1685

class fake_pydbus_raises_exception_on_connect:

1686

"""fake dbus-python module"""

1687

@classmethod

1688

def SystemBus(cls):

1689

def get(busname, objectpath):

1690

raise gi.repository.GLib.Error()

1691

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

1692

return Bus(get=get)

1693

1694

def test_set_property_uses_setattr(self):

1695

class Object:

1696

pass

1697

obj = Object()

1698

class pydbus_spy:

1699

class SystemBus:

1700

@staticmethod

1701

def get(busname, objectpath):

1702

return {"interface": obj}

1703

bus = pydbus_adapter.SystemBus(pydbus_spy)

1704

value = Unique()

1705

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

1706

value)

1707

self.assertIs(value, obj.key)

1708

1709

def test_get_suppresses_xml_deprecation_warning(self):

1710

if sys.version_info.major >= 3:

1711

return

1712

class stub_pydbus_get:

1713

class SystemBus:

1714

@staticmethod

1715

def get(busname, objectpath):

1716

warnings.warn_explicit(

1717

"deprecated", DeprecationWarning,

1718

"xml.etree.ElementTree", 0)

1719

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1720

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

1721

warnings.simplefilter("always")

1722

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

1723

self.assertEqual(0, len(w))

1724

1725

1726

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1727

class stub_pydbus:

1728

"""stub pydbus module"""

1729

class SystemBus:

1730

@staticmethod

1731

def get(busname, objectpath):

1732

return Unique()

1733

1734

def setUp(self):

1735

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

1736

1737

def test_returns_distinct_objectpaths(self):

1738

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

1739

self.assertIsInstance(obj1, Unique)

1740

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

1741

self.assertIsInstance(obj2, Unique)

1742

self.assertIsNot(obj1, obj2)

1743

1744

def test_returns_distinct_busnames(self):

1745

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

1746

self.assertIsInstance(obj1, Unique)

1747

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

1748

self.assertIsInstance(obj2, Unique)

1749

self.assertIsNot(obj1, obj2)

1750

1751

def test_returns_distinct_both(self):

1752

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

1753

self.assertIsInstance(obj1, Unique)

1754

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

1755

self.assertIsInstance(obj2, Unique)

1756

self.assertIsNot(obj1, obj2)

1757

1758

def test_returns_same(self):

1759

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

1760

self.assertIsInstance(obj1, Unique)

1761

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

1762

self.assertIsInstance(obj2, Unique)

1763

self.assertIs(obj1, obj2)

1764

1765

def test_returns_same_old(self):

1766

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

1767

self.assertIsInstance(obj1, Unique)

1768

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

1769

self.assertIsInstance(obj2, Unique)

1770

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

1771

self.assertIsInstance(obj1b, Unique)

1772

self.assertIsNot(obj1, obj2)

1773

self.assertIsNot(obj2, obj1b)

1774

self.assertIs(obj1, obj1b)

1775

1776

1777

1519

class Test_commands_from_options(unittest.TestCase):

1778

1520

1779

1521

def setUp(self):

1784

1526

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

1785

1527

command.IsEnabled)

1786

1528

1787

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

1788

clients=None, **cmd_attrs):

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

1789

1531

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

1790

1532

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

1791

1533

options = self.parser.parse_args(args)

1792

1534

check_option_syntax(self.parser, options)

1793

1535

commands = commands_from_options(options)

1794

self.assertEqual(length, len(commands))

1795

for command in commands:

1796

if isinstance(command, command_cls):

1797

break

1798

else:

1799

self.assertIsInstance(command, command_cls)

1800

if clients is not None:

1801

self.assertEqual(clients, options.client)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

1802

1539

for key, value in cmd_attrs.items():

1803

1540

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

1804

1541

1805

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

1806

for cmd in commands:

1807

self.assert_command_from_args(args, cmd,

1808

length=len(commands),

1809

clients=clients)

1810

1811

1542

def test_is_enabled_short(self):

1812

1543

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

1813

1544

command.IsEnabled)

2004

1735

verbose=True)

2005

1736

2006

1737

2007

def test_manual_page_example_1(self):

2008

self.assert_command_from_args("",

2009

command.PrintTable,

2010

clients=[],

2011

verbose=False)

2012

2013

def test_manual_page_example_2(self):

2014

self.assert_command_from_args(

2015

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

2016

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

2017

"foo2.example.org"],

2018

verbose=True)

2019

2020

def test_manual_page_example_3(self):

2021

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

2022

command.Enable,

2023

clients=[])

2024

2025

def test_manual_page_example_4(self):

2026

self.assert_commands_from_args(

2027

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

2028

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

2029

[command.SetTimeout, command.SetInterval],

2030

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

2031

2032

def test_manual_page_example_5(self):

2033

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

2034

command.Approve,

2035

clients=[])

2036

2037

2038

1738

class TestCommand(unittest.TestCase):

2039

1739

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

2040

1740

2370

2070

def runTest(self):

2371

2071

if not hasattr(self, "command"):

2372

2072

return # Abstract TestCase class

2373

2374

if hasattr(self, "values_to_set"):

2375

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

2376

values_to_get = getattr(self, "values_to_get",

2377

self.values_to_set)

2378

else:

2379

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

2380

values_to_get = self.values_to_get

2381

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

2382

for clientpath in self.bus.clients:

2383

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

2384

Unique())

2385

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

2386

for clientpath in self.bus.clients:

2387

value = (self.bus.clients[clientpath]

2388

[self.propname])

2073

values_to_get = getattr(self, "values_to_get",

2074

self.values_to_set)

2075

for value_to_set, value_to_get in zip(self.values_to_set,

2076

values_to_get):

2077

for clientpath in self.bus.clients:

2078

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

2079

self.run_command(value_to_set, self.bus.clients)

2080

for clientpath in self.bus.clients:

2081

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

2389

2082

self.assertNotIsInstance(value, Unique)

2390

2083

self.assertEqual(value_to_get, value)

2391

2084

2085

def run_command(self, value, clients):

2086

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

2087

2392

2088

2393

2089

class TestEnableCmd(TestPropertySetterCmd):

2394

2090

command = command.Enable

2395

2091

propname = "Enabled"

2396

values_to_get = [True]

2092

values_to_set = [True]

2397

2093

2398

2094

2399

2095

class TestDisableCmd(TestPropertySetterCmd):

2400

2096

command = command.Disable

2401

2097

propname = "Enabled"

2402

values_to_get = [False]

2098

values_to_set = [False]

2403

2099

2404

2100

2405

2101

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2406

2102

command = command.BumpTimeout

2407

2103

propname = "LastCheckedOK"

2408

values_to_get = [""]

2104

values_to_set = [""]

2409

2105

2410

2106

2411

2107

class TestStartCheckerCmd(TestPropertySetterCmd):

2412

2108

command = command.StartChecker

2413

2109

propname = "CheckerRunning"

2414

values_to_get = [True]

2110

values_to_set = [True]

2415

2111

2416

2112

2417

2113

class TestStopCheckerCmd(TestPropertySetterCmd):

2418

2114

command = command.StopChecker

2419

2115

propname = "CheckerRunning"

2420

values_to_get = [False]

2116

values_to_set = [False]

2421

2117

2422

2118

2423

2119

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2424

2120

command = command.ApproveByDefault

2425

2121

propname = "ApprovedByDefault"

2426

values_to_get = [True]

2122

values_to_set = [True]

2427

2123

2428

2124

2429

2125

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2430

2126

command = command.DenyByDefault

2431

2127

propname = "ApprovedByDefault"

2432

values_to_get = [False]

2433

2434

2435

class TestSetCheckerCmd(TestPropertySetterCmd):

2128

values_to_set = [False]

2129

2130

2131

class TestPropertySetterValueCmd(TestPropertySetterCmd):

2132

"""Abstract class for tests of PropertySetterValueCmd classes"""

2133

2134

def run_command(self, value, clients):

2135

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

2136

2137

2138

class TestSetCheckerCmd(TestPropertySetterValueCmd):

2436

2139

command = command.SetChecker

2437

2140

propname = "Checker"

2438

2141

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

2439

2142

2440

2143

2441

class TestSetHostCmd(TestPropertySetterCmd):

2144

class TestSetHostCmd(TestPropertySetterValueCmd):

2442

2145

command = command.SetHost

2443

2146

propname = "Host"

2444

2147

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

2445

2148

2446

2149

2447

class TestSetSecretCmd(TestPropertySetterCmd):

2150

class TestSetSecretCmd(TestPropertySetterValueCmd):

2448

2151

command = command.SetSecret

2449

2152

propname = "Secret"

2450

2153

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

2452

2155

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

2453

2156

2454

2157

2455

class TestSetTimeoutCmd(TestPropertySetterCmd):

2158

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

2456

2159

command = command.SetTimeout

2457

2160

propname = "Timeout"

2458

2161

values_to_set = [datetime.timedelta(),

2463

2166

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

2464

2167

2465

2168

2466

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2169

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

2467

2170

command = command.SetExtendedTimeout

2468

2171

propname = "ExtendedTimeout"

2469

2172

values_to_set = [datetime.timedelta(),

2474

2177

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

2475

2178

2476

2179

2477

class TestSetIntervalCmd(TestPropertySetterCmd):

2180

class TestSetIntervalCmd(TestPropertySetterValueCmd):

2478

2181

command = command.SetInterval

2479

2182

propname = "Interval"

2480

2183

values_to_set = [datetime.timedelta(),

2485

2188

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

2486

2189

2487

2190

2488

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2191

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

2489

2192

command = command.SetApprovalDelay

2490

2193

propname = "ApprovalDelay"

2491

2194

values_to_set = [datetime.timedelta(),

2496

2199

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

2497

2200

2498

2201

2499

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2202

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

2500

2203

command = command.SetApprovalDuration

2501

2204

propname = "ApprovalDuration"

2502

2205

values_to_set = [datetime.timedelta(),