/mandos/trunk : revision 237.4.124

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*")))); -*-

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 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

45

import io

46

import tempfile

47

import contextlib

48

import abc

49

50

import dbus as dbus_python

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

51

65

52

66

# Show warnings by default

53

67

if not sys.warnoptions:

67

81

68

82

locale.setlocale(locale.LC_ALL, "")

69

83

70

version = "1.8.3"

84

version = "1.8.9"

71

85

72

86

73

87

def main():

82

96

if options.debug:

83

97

log.setLevel(logging.DEBUG)

84

98

85

bus = dbus_python_adapter.CachingBus(dbus_python)

99

if pydbus is not None:

100

bus = pydbus_adapter.CachingBus(pydbus)

101

else:

102

bus = dbus_python_adapter.CachingBus(dbus_python)

86

103

87

104

try:

88

105

all_clients = bus.get_clients_and_properties()

122

139

help="Select all clients")

123

140

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

124

141

help="Print all fields")

125

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

142

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

143

action="append_const", default=[],

144

const=command.DumpJSON(),

126

145

help="Dump client data in JSON format")

127

146

enable_disable = parser.add_mutually_exclusive_group()

128

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

147

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

148

action="append_const", default=[],

149

const=command.Enable(),

129

150

help="Enable client")

130

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

131

action="store_true",

151

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

152

action="append_const", default=[],

153

const=command.Disable(),

132

154

help="disable client")

133

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

155

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

156

action="append_const", default=[],

157

const=command.BumpTimeout(),

134

158

help="Bump timeout for client")

135

159

start_stop_checker = parser.add_mutually_exclusive_group()

136

160

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

137

action="store_true",

161

dest="commands",

162

action="append_const", default=[],

163

const=command.StartChecker(),

138

164

help="Start checker for client")

139

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

140

action="store_true",

165

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

166

action="append_const", default=[],

167

const=command.StopChecker(),

141

168

help="Stop checker for client")

142

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

169

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

170

action="append_const", default=[],

171

const=command.IsEnabled(),

143

172

help="Check if client is enabled")

144

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

173

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

174

action="append_const", default=[],

175

const=command.Remove(),

145

176

help="Remove client")

146

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

177

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

178

action="append", default=[],

179

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

147

180

help="Set checker command 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")

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")

154

196

approve_deny_default = parser.add_mutually_exclusive_group()

155

197

approve_deny_default.add_argument(

156

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

157

default=None, dest="approved_by_default",

198

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

199

action="append_const", default=[],

200

const=command.ApproveByDefault(),

158

201

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

159

202

approve_deny_default.add_argument(

160

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

161

dest="approved_by_default",

203

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

204

action="append_const", default=[],

205

const=command.DenyByDefault(),

162

206

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

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")

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")

171

226

approve_deny = parser.add_mutually_exclusive_group()

172

227

approve_deny.add_argument(

173

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

228

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

229

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

174

230

help="Approve any current client request")

175

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

231

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

232

action="append_const", default=[],

233

const=command.Deny(),

176

234

help="Deny any current client request")

177

235

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

178

236

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

195

253

def rfc3339_duration_to_delta(duration):

196

254

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

197

255

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)

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

214

273

>>> # Can not be empty:

215

274

>>> rfc3339_duration_to_delta("")

216

275

Traceback (most recent call last):

326

385

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

327

386

and return a datetime.timedelta

328

387

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)

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

343

402

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

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

403

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

404

True

346

405

347

406

"""

348

407

369

428

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

370

429

argparse"""

371

430

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))

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)

391

454

392

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

455

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

393

456

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

394

if options.verbose and has_actions(options):

457

if options.verbose and has_commands(options):

395

458

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

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

459

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

460

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

398

461

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

399

if options.all and not has_actions(options):

462

if options.all and not has_commands(options):

400

463

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

401

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

464

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

465

and len(options.client) > 1):

402

466

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

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):

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:

414

476

415

477

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

416

478

def get_managed_objects(self, busname, objectpath):

462

524

pass

463

525

464

526

465

class dbus_python_adapter(object):

527

class dbus_python_adapter:

466

528

467

529

class SystemBus(dbus.MandosBus):

468

530

"""Use dbus-python"""

513

575

for key, subval in value.items()}

514

576

return value

515

577

578

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

579

if key == "Secret":

580

if not isinstance(value, bytes):

581

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

582

value = self.dbus_python.ByteArray(value)

583

return self.set_property(self.busname, objectpath,

584

self.client_interface, key,

585

value)

516

586

517

class SilenceLogger(object):

587

class SilenceLogger:

518

588

"Simple context manager to silence a particular logger"

519

589

def __init__(self, loggername):

520

590

self.logger = logging.getLogger(loggername)

549

619

return new_object

550

620

551

621

622

class pydbus_adapter:

623

class SystemBus(dbus.MandosBus):

624

def __init__(self, module=pydbus):

625

self.pydbus = module

626

self.bus = self.pydbus.SystemBus()

627

628

@contextlib.contextmanager

629

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

630

try:

631

yield

632

except gi.repository.GLib.Error as e:

633

# This does what "raise from" would do

634

exc = exception_class(*e.args)

635

exc.__cause__ = e

636

raise exc

637

638

def call_method(self, methodname, busname, objectpath,

639

interface, *args):

640

proxy_object = self.get(busname, objectpath)

641

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

642

interface, methodname,

643

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

644

method = getattr(proxy_object[interface], methodname)

645

with self.convert_exception():

646

return method(*args)

647

648

def get(self, busname, objectpath):

649

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

650

busname, objectpath)

651

with self.convert_exception(dbus.ConnectFailed):

652

if sys.version_info.major <= 2:

653

with warnings.catch_warnings():

654

warnings.filterwarnings(

655

"ignore", "", DeprecationWarning,

656

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

657

return self.bus.get(busname, objectpath)

658

else:

659

return self.bus.get(busname, objectpath)

660

661

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

662

value):

663

proxy_object = self.get(busname, objectpath)

664

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

665

objectpath, self.properties_iface, interface,

666

key, value)

667

setattr(proxy_object[interface], key, value)

668

669

class CachingBus(SystemBus):

670

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

671

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

672

self.object_cache = {}

673

super(pydbus_adapter.CachingBus,

674

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

675

def get(self, busname, objectpath):

676

try:

677

return self.object_cache[(busname, objectpath)]

678

except KeyError:

679

new_object = (super(pydbus_adapter.CachingBus, self)

680

.get(busname, objectpath))

681

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

682

return new_object

683

684

552

685

def commands_from_options(options):

553

686

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())

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)

589

705

else:

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))

706

cleaned_after.append(cmd)

707

if cleaned_after != after_remove:

708

commands = before_remove + cleaned_after

618

709

619

710

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

620

711

# optionally verbosely

624

715

return commands

625

716

626

717

627

class command(object):

718

class command:

628

719

"""A namespace for command classes"""

629

720

630

class Base(object):

721

class Base:

631

722

"""Abstract base class for commands"""

632

723

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

633

724

"""Normal commands should implement run_on_one_client(),

696

787

keywords = self.all_keywords

697

788

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

698

789

699

class TableOfClients(object):

790

class TableOfClients:

700

791

tableheaders = {

701

792

"Name": "Name",

702

793

"Enabled": "Enabled",

835

926

def __init__(self, value):

836

927

self.value_to_set = value

837

928

929

@classmethod

930

def argparse(cls, argtype):

931

def cmdtype(arg):

932

return cls(argtype(arg))

933

return cmdtype

838

934

839

935

class SetChecker(PropertySetterValue):

840

936

propname = "Checker"

928

1024

"output"))

929

1025

930

1026

931

class Unique(object):

1027

class Unique:

932

1028

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

933

1029

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

934

1030

966

1062

967

1063

def test_actions_requires_client_or_all(self):

968

1064

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

969

options = self.parser.parse_args()

970

setattr(options, action, value)

1065

args = self.actionargs(action, value)

971

1066

with self.assertParseError():

972

self.check_option_syntax(options)

1067

self.parse_args(args)

973

1068

974

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

1069

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

975

1070

# check_option_syntax()

976

1071

actions = {

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,

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,

998

1091

}

999

1092

1093

@staticmethod

1094

def actionargs(action, value, *args):

1095

if value is not None:

1096

return [action, value] + list(args)

1097

else:

1098

return [action] + list(args)

1099

1000

1100

@contextlib.contextmanager

1001

1101

def assertParseError(self):

1002

1102

with self.assertRaises(SystemExit) as e:

1007

1107

# /argparse.html#exiting-methods

1008

1108

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

1009

1109

1110

def parse_args(self, args):

1111

options = self.parser.parse_args(args)

1112

check_option_syntax(self.parser, options)

1113

1010

1114

@staticmethod

1011

1115

@contextlib.contextmanager

1012

1116

def redirect_stderr_to_devnull():

1023

1127

1024

1128

def test_actions_all_conflicts_with_verbose(self):

1025

1129

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

1026

options = self.parser.parse_args()

1027

setattr(options, action, value)

1028

options.all = True

1029

options.verbose = True

1130

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

1131

"--verbose")

1030

1132

with self.assertParseError():

1031

self.check_option_syntax(options)

1133

self.parse_args(args)

1032

1134

1033

1135

def test_actions_with_client_conflicts_with_verbose(self):

1034

1136

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

1035

options = self.parser.parse_args()

1036

setattr(options, action, value)

1037

options.verbose = True

1038

options.client = ["client"]

1137

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

1138

"client")

1039

1139

with self.assertParseError():

1040

self.check_option_syntax(options)

1140

self.parse_args(args)

1041

1141

1042

1142

def test_dump_json_conflicts_with_verbose(self):

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1143

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

1046

1144

with self.assertParseError():

1047

self.check_option_syntax(options)

1145

self.parse_args(args)

1048

1146

1049

1147

def test_dump_json_conflicts_with_action(self):

1050

1148

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

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1149

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

1054

1150

with self.assertParseError():

1055

self.check_option_syntax(options)

1151

self.parse_args(args)

1056

1152

1057

1153

def test_all_can_not_be_alone(self):

1058

options = self.parser.parse_args()

1059

options.all = True

1154

args = ["--all"]

1060

1155

with self.assertParseError():

1061

self.check_option_syntax(options)

1156

self.parse_args(args)

1062

1157

1063

1158

def test_all_is_ok_with_any_action(self):

1064

1159

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

1065

options = self.parser.parse_args()

1066

setattr(options, action, value)

1067

options.all = True

1068

self.check_option_syntax(options)

1160

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

1161

self.parse_args(args)

1069

1162

1070

1163

def test_any_action_is_ok_with_one_client(self):

1071

1164

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

1072

options = self.parser.parse_args()

1073

setattr(options, action, value)

1074

options.client = ["client"]

1075

self.check_option_syntax(options)

1165

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

1166

self.parse_args(args)

1076

1167

1077

1168

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1079

1169

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

1080

if action == "is_enabled":

1170

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

1081

1171

continue

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1172

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

1173

self.parse_args(args)

1085

1174

1086

1175

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1088

1176

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

1089

if action == "is_enabled":

1177

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

1090

1178

continue

1091

setattr(options, action, value)

1092

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

1093

self.check_option_syntax(options)

1179

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

1180

"client2")

1181

self.parse_args(args)

1094

1182

1095

1183

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1096

1184

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

1097

if action == "is_enabled":

1185

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

1098

1186

continue

1099

options = self.parser.parse_args()

1100

setattr(options, action, value)

1101

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

1102

self.check_option_syntax(options)

1187

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

1188

"client2")

1189

self.parse_args(args)

1103

1190

1104

1191

def test_is_enabled_fails_without_client(self):

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1192

args = ["--is-enabled"]

1107

1193

with self.assertParseError():

1108

self.check_option_syntax(options)

1194

self.parse_args(args)

1109

1195

1110

1196

def test_is_enabled_fails_with_two_clients(self):

1111

options = self.parser.parse_args()

1112

options.is_enabled = True

1113

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

1197

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

1114

1198

with self.assertParseError():

1115

self.check_option_syntax(options)

1199

self.parse_args(args)

1116

1200

1117

1201

def test_remove_can_only_be_combined_with_action_deny(self):

1118

1202

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

1119

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

1203

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

1120

1204

continue

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1205

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

1206

"--remove")

1125

1207

with self.assertParseError():

1126

self.check_option_syntax(options)

1208

self.parse_args(args)

1127

1209

1128

1210

1129

1211

class Test_dbus_exceptions(unittest.TestCase):

1232

1314

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1233

1315

1234

1316

def MockDBusPython_func(self, func):

1235

class mock_dbus_python(object):

1317

class mock_dbus_python:

1236

1318

"""mock dbus-python module"""

1237

class exceptions(object):

1319

class exceptions:

1238

1320

"""Pseudo-namespace"""

1239

1321

class DBusException(Exception):

1240

1322

pass

1241

class SystemBus(object):

1323

class SystemBus:

1242

1324

@staticmethod

1243

1325

def get_object(busname, objectpath):

1244

1326

DBusObject = collections.namedtuple(

1245

"DBusObject", ("methodname",))

1327

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

1246

1328

def method(*args, **kwargs):

1247

1329

self.assertEqual({"dbus_interface":

1248

1330

"interface"},

1249

1331

kwargs)

1250

1332

return func(*args)

1251

return DBusObject(methodname=method)

1252

class Boolean(object):

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:

1253

1344

def __init__(self, value):

1254

1345

self.value = bool(value)

1255

1346

def __bool__(self):

1260

1351

pass

1261

1352

class Dictionary(dict):

1262

1353

pass

1354

class ByteArray(bytes):

1355

pass

1263

1356

return mock_dbus_python

1264

1357

1265

1358

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

1442

1535

# Make sure the dbus logger was suppressed

1443

1536

self.assertEqual(0, counting_handler.count)

1444

1537

1538

def test_Set_Secret_sends_bytearray(self):

1539

ret = [None]

1540

def func(*args, **kwargs):

1541

ret[0] = (args, kwargs)

1542

mock_dbus_python = self.MockDBusPython_func(func)

1543

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1544

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

1545

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

1546

mock_dbus_python.ByteArray(b"value")),

1547

{"dbus_interface":

1548

"org.freedesktop.DBus.Properties"})

1549

self.assertEqual(expected_call, ret[0])

1550

if sys.version_info.major == 2:

1551

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

1552

mock_dbus_python.ByteArray)

1553

1445

1554

def test_get_object_converts_to_correct_exception(self):

1446

1555

bus = dbus_python_adapter.SystemBus(

1447

1556

self.fake_dbus_python_raises_exception_on_connect)

1449

1558

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

1450

1559

"objectpath", "interface")

1451

1560

1452

class fake_dbus_python_raises_exception_on_connect(object):

1561

class fake_dbus_python_raises_exception_on_connect:

1453

1562

"""fake dbus-python module"""

1454

class exceptions(object):

1563

class exceptions:

1455

1564

"""Pseudo-namespace"""

1456

1565

class DBusException(Exception):

1457

1566

pass

1465

1574

1466

1575

1467

1576

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1468

class mock_dbus_python(object):

1577

class mock_dbus_python:

1469

1578

"""mock dbus-python modules"""

1470

class SystemBus(object):

1579

class SystemBus:

1471

1580

@staticmethod

1472

1581

def get_object(busname, objectpath):

1473

1582

return Unique()

1516

1625

self.assertIs(obj1, obj1b)

1517

1626

1518

1627

1628

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1629

1630

def Stub_pydbus_func(self, func):

1631

class stub_pydbus:

1632

"""stub pydbus module"""

1633

class SystemBus:

1634

@staticmethod

1635

def get(busname, objectpath):

1636

DBusObject = collections.namedtuple(

1637

"DBusObject", ("methodname",))

1638

return {"interface":

1639

DBusObject(methodname=func)}

1640

return stub_pydbus

1641

1642

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

1643

interface, *args):

1644

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

1645

return bus.call_method(methodname, busname, objectpath,

1646

interface, *args)

1647

1648

def test_call_method_returns(self):

1649

expected_method_return = Unique()

1650

method_args = (Unique(), Unique())

1651

def func(*args):

1652

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

1653

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

1654

self.assertIs(marg, arg)

1655

return expected_method_return

1656

stub_pydbus = self.Stub_pydbus_func(func)

1657

bus = pydbus_adapter.SystemBus(stub_pydbus)

1658

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

1659

"objectpath", "interface",

1660

*method_args)

1661

self.assertIs(ret, expected_method_return)

1662

1663

def test_call_method_handles_exception(self):

1664

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

1665

1666

def func():

1667

raise gi.repository.GLib.Error()

1668

1669

stub_pydbus = self.Stub_pydbus_func(func)

1670

bus = pydbus_adapter.SystemBus(stub_pydbus)

1671

1672

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

1673

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

1674

"objectpath", "interface")

1675

1676

self.assertNotIsInstance(e, dbus.ConnectFailed)

1677

1678

def test_get_converts_to_correct_exception(self):

1679

bus = pydbus_adapter.SystemBus(

1680

self.fake_pydbus_raises_exception_on_connect)

1681

with self.assertRaises(dbus.ConnectFailed):

1682

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

1683

"objectpath", "interface")

1684

1685

class fake_pydbus_raises_exception_on_connect:

1686

"""fake dbus-python module"""

1687

@classmethod

1688

def SystemBus(cls):

1689

def get(busname, objectpath):

1690

raise gi.repository.GLib.Error()

1691

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

1692

return Bus(get=get)

1693

1694

def test_set_property_uses_setattr(self):

1695

class Object:

1696

pass

1697

obj = Object()

1698

class pydbus_spy:

1699

class SystemBus:

1700

@staticmethod

1701

def get(busname, objectpath):

1702

return {"interface": obj}

1703

bus = pydbus_adapter.SystemBus(pydbus_spy)

1704

value = Unique()

1705

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

1706

value)

1707

self.assertIs(value, obj.key)

1708

1709

def test_get_suppresses_xml_deprecation_warning(self):

1710

if sys.version_info.major >= 3:

1711

return

1712

class stub_pydbus_get:

1713

class SystemBus:

1714

@staticmethod

1715

def get(busname, objectpath):

1716

warnings.warn_explicit(

1717

"deprecated", DeprecationWarning,

1718

"xml.etree.ElementTree", 0)

1719

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1720

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

1721

warnings.simplefilter("always")

1722

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

1723

self.assertEqual(0, len(w))

1724

1725

1726

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1727

class stub_pydbus:

1728

"""stub pydbus module"""

1729

class SystemBus:

1730

@staticmethod

1731

def get(busname, objectpath):

1732

return Unique()

1733

1734

def setUp(self):

1735

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

1736

1737

def test_returns_distinct_objectpaths(self):

1738

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

1739

self.assertIsInstance(obj1, Unique)

1740

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

1741

self.assertIsInstance(obj2, Unique)

1742

self.assertIsNot(obj1, obj2)

1743

1744

def test_returns_distinct_busnames(self):

1745

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

1746

self.assertIsInstance(obj1, Unique)

1747

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

1748

self.assertIsInstance(obj2, Unique)

1749

self.assertIsNot(obj1, obj2)

1750

1751

def test_returns_distinct_both(self):

1752

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

1753

self.assertIsInstance(obj1, Unique)

1754

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

1755

self.assertIsInstance(obj2, Unique)

1756

self.assertIsNot(obj1, obj2)

1757

1758

def test_returns_same(self):

1759

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

1760

self.assertIsInstance(obj1, Unique)

1761

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

1762

self.assertIsInstance(obj2, Unique)

1763

self.assertIs(obj1, obj2)

1764

1765

def test_returns_same_old(self):

1766

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

1767

self.assertIsInstance(obj1, Unique)

1768

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

1769

self.assertIsInstance(obj2, Unique)

1770

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

1771

self.assertIsInstance(obj1b, Unique)

1772

self.assertIsNot(obj1, obj2)

1773

self.assertIsNot(obj2, obj1b)

1774

self.assertIs(obj1, obj1b)

1775

1776

1519

1777

class Test_commands_from_options(unittest.TestCase):

1520

1778

1521

1779

def setUp(self):

1526

1784

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

1527

1785

command.IsEnabled)

1528

1786

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

1787

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

1788

clients=None, **cmd_attrs):

1531

1789

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

1532

1790

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

1533

1791

options = self.parser.parse_args(args)

1534

1792

check_option_syntax(self.parser, options)

1535

1793

commands = commands_from_options(options)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

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)

1539

1802

for key, value in cmd_attrs.items():

1540

1803

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

1541

1804

1805

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

1806

for cmd in commands:

1807

self.assert_command_from_args(args, cmd,

1808

length=len(commands),

1809

clients=clients)

1810

1542

1811

def test_is_enabled_short(self):

1543

1812

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

1544

1813

command.IsEnabled)

1735

2004

verbose=True)

1736

2005

1737

2006

2007

def test_manual_page_example_1(self):

2008

self.assert_command_from_args("",

2009

command.PrintTable,

2010

clients=[],

2011

verbose=False)

2012

2013

def test_manual_page_example_2(self):

2014

self.assert_command_from_args(

2015

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

2016

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

2017

"foo2.example.org"],

2018

verbose=True)

2019

2020

def test_manual_page_example_3(self):

2021

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

2022

command.Enable,

2023

clients=[])

2024

2025

def test_manual_page_example_4(self):

2026

self.assert_commands_from_args(

2027

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

2028

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

2029

[command.SetTimeout, command.SetInterval],

2030

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

2031

2032

def test_manual_page_example_5(self):

2033

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

2034

command.Approve,

2035

clients=[])

2036

2037

1738

2038

class TestCommand(unittest.TestCase):

1739

2039

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

1740

2040