/mandos/trunk : revision 1040

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

47

49

48

import dbus

50

49

64

63

65

64

locale.setlocale(locale.LC_ALL, "")

66

65

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

66

domain = "se.recompile"

67

busname = domain + ".Mandos"

68

server_path = "/"

69

server_interface = domain + ".Mandos"

70

client_interface = domain + ".Mandos.Client"

73

71

version = "1.8.3"

74

72

75

73

277

275

# Abstract classes first

278

276

class Command(object):

279

277

"""Abstract class for commands"""

280

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

278

def run(self, mandos, clients):

281

279

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

282

280

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

283

281

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

284

282

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)

283

for client, properties in clients.items():

289

284

self.run_on_one_client(client, properties)

290

285

291

286

class PrintCmd(Command):

297

292

"LastApprovalRequest", "ApprovalDelay",

298

293

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

299

294

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

295

def run(self, mandos, clients):

296

print(self.output(clients))

304

297

305

298

class PropertyCmd(Command):

306

299

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

307

300

def run_on_one_client(self, client, properties):

308

301

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

302

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

317

303

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

321

304

322

class PropertyValueCmd(PropertyCmd):

323

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

305

class ValueArgumentMixIn(object):

306

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

324

307

def __init__(self, value):

325

308

self.value_to_set = value

326

309

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

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

329

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

310

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

311

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

312

milliseconds."""

330

313

@property

331

314

def value_to_set(self):

332

315

return self._vts

333

316

@value_to_set.setter

334

317

def value_to_set(self, value):

335

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

318

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

336

319

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

337

320

338

321

# Actual (non-abstract) command classes

342

325

self.verbose = verbose

343

326

344

327

def output(self, clients):

345

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

346

"LastCheckedOK")

328

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

347

329

keywords = default_keywords

348

330

if self.verbose:

349

331

keywords = self.all_keywords

350

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

332

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

351

333

352

334

class TableOfClients(object):

353

335

tableheaders = {

439

421

return value

440

422

441

423

class IsEnabledCmd(Command):

442

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

443

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

424

def run_on_one_client(self, client, properties):

444

425

if self.is_enabled(client, properties):

445

426

sys.exit(0)

446

427

sys.exit(1)

447

428

def is_enabled(self, client, properties):

448

return properties["Enabled"]

429

return bool(properties["Enabled"])

449

430

450

431

class RemoveCmd(Command):

451

432

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

433

self.mandos.RemoveClient(client.__dbus_object_path__)

456

434

457

435

class ApproveCmd(Command):

458

436

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

437

client.Approve(dbus.Boolean(True),

462

dbus_interface=client_dbus_interface)

438

dbus_interface=client_interface)

463

439

464

440

class DenyCmd(Command):

465

441

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

442

client.Approve(dbus.Boolean(False),

469

dbus_interface=client_dbus_interface)

443

dbus_interface=client_interface)

470

444

471

445

class EnableCmd(PropertyCmd):

472

propname = "Enabled"

446

property = "Enabled"

473

447

value_to_set = dbus.Boolean(True)

474

448

475

449

class DisableCmd(PropertyCmd):

476

propname = "Enabled"

450

property = "Enabled"

477

451

value_to_set = dbus.Boolean(False)

478

452

479

453

class BumpTimeoutCmd(PropertyCmd):

480

propname = "LastCheckedOK"

454

property = "LastCheckedOK"

481

455

value_to_set = ""

482

456

483

457

class StartCheckerCmd(PropertyCmd):

484

propname = "CheckerRunning"

458

property = "CheckerRunning"

485

459

value_to_set = dbus.Boolean(True)

486

460

487

461

class StopCheckerCmd(PropertyCmd):

488

propname = "CheckerRunning"

462

property = "CheckerRunning"

489

463

value_to_set = dbus.Boolean(False)

490

464

491

465

class ApproveByDefaultCmd(PropertyCmd):

492

propname = "ApprovedByDefault"

466

property = "ApprovedByDefault"

493

467

value_to_set = dbus.Boolean(True)

494

468

495

469

class DenyByDefaultCmd(PropertyCmd):

496

propname = "ApprovedByDefault"

470

property = "ApprovedByDefault"

497

471

value_to_set = dbus.Boolean(False)

498

472

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

473

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

474

property = "Checker"

475

476

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

477

property = "Host"

478

479

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

507

480

@property

508

481

def value_to_set(self):

509

482

return self._vts

512

485

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

513

486

self._vts = value.read()

514

487

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"

488

property = "Secret"

489

490

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

491

property = "Timeout"

492

493

class SetExtendedTimeoutCmd(PropertyCmd,

494

MillisecondsValueArgumentMixIn):

495

property = "ExtendedTimeout"

496

497

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

498

property = "Interval"

499

500

class SetApprovalDelayCmd(PropertyCmd,

501

MillisecondsValueArgumentMixIn):

502

property = "ApprovalDelay"

503

504

class SetApprovalDurationCmd(PropertyCmd,

505

MillisecondsValueArgumentMixIn):

506

property = "ApprovalDuration"

530

507

531

508

def add_command_line_options(parser):

532

509

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

588

565

help="Approve any current client request")

589

566

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

590

567

help="Deny any current client request")

591

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

592

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

593

568

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

594

569

help="Run self-test")

595

570

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

620

595

if options.is_enabled:

621

596

commands.append(IsEnabledCmd())

622

597

598

if options.remove:

599

commands.append(RemoveCmd())

600

623

601

if options.checker is not None:

624

602

commands.append(SetCheckerCmd(options.checker))

625

603

658

636

if options.deny:

659

637

commands.append(DenyCmd())

660

638

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

639

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

665

640

# optionally verbosely

666

641

if not commands:

670

645

671

646

672

647

def check_option_syntax(parser, options):

673

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

674

argparse"""

675

648

676

649

def has_actions(options):

677

650

return any((options.enable,

704

677

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

705

678

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

706

679

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

680

713

681

714

682

def main():

722

690

723

691

clientnames = options.client

724

692

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

693

try:

729

694

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)

695

mandos_dbus_objc = bus.get_object(busname, server_path)

734

696

except dbus.exceptions.DBusException:

735

697

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

736

698

sys.exit(1)

737

699

738

700

mandos_serv = dbus.Interface(mandos_dbus_objc,

739

dbus_interface=server_dbus_interface)

701

dbus_interface=server_interface)

740

702

mandos_serv_object_manager = dbus.Interface(

741

703

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

742

704

748

710

dbus_filter = NullFilter()

749

711

try:

750

712

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]

713

mandos_clients = {path: ifs_and_props[client_interface]

754

714

for path, ifs_and_props in

755

715

mandos_serv_object_manager

756

716

.GetManagedObjects().items()

757

if client_dbus_interface in ifs_and_props}

717

if client_interface in ifs_and_props}

758

718

except dbus.exceptions.DBusException as e:

759

719

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

760

720

"\n%s", e)

767

727

clients = {}

768

728

769

729

if not clientnames:

770

clients = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

730

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

731

for path, properties in mandos_clients.items()}

772

732

else:

773

733

for name in clientnames:

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

734

for path, client in mandos_clients.items():

735

if client["Name"] == name:

736

client_objc = bus.get_object(busname, path)

737

clients[client_objc] = client

777

738

break

778

739

else:

779

740

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

782

743

# Run all commands on clients

783

744

commands = commands_from_options(options)

784

745

for command in commands:

785

command.run(clients, bus, mandos_serv)

746

command.run(mandos_serv, clients)

786

747

787

748

788

749

class Test_milliseconds_to_string(unittest.TestCase):

837

798

testcase = self

838

799

class MockClient(object):

839

800

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

840

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

801

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

841

802

self.attributes = attributes

842

803

self.attributes["Name"] = name

843

804

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]

805

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

806

testcase.assertEqual(interface, client_interface)

807

testcase.assertEqual(dbus_interface,

808

dbus.PROPERTIES_IFACE)

809

self.attributes[property] = value

810

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

811

testcase.assertEqual(interface, client_interface)

812

testcase.assertEqual(dbus_interface,

813

dbus.PROPERTIES_IFACE)

814

return self.attributes[property]

854

815

def Approve(self, approve, dbus_interface):

855

testcase.assertEqual(dbus_interface,

856

client_dbus_interface)

816

testcase.assertEqual(dbus_interface, client_interface)

857

817

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

858

818

dbus_interface)))

859

819

self.client = MockClient(

906

866

LastCheckerStatus=-2)

907

867

self.clients = collections.OrderedDict(

908

868

[

909

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

910

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

869

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

870

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

911

871

])

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

872

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

924

873

925

874

class TestPrintTableCmd(TestCmd):

926

875

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

))

876

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

877

expected_output = """

878

Name Enabled Timeout Last Successful Check

879

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

880

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

881

"""[1:-1]

933

882

self.assertEqual(output, expected_output)

934

883

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

))

884

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

885

expected_output = """

886

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

887

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

888

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

889

"""[1:-1]

966

890

self.assertEqual(output, expected_output)

967

891

def test_one_client(self):

968

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

892

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

969

893

expected_output = """

970

894

Name Enabled Timeout Last Successful Check

971

895

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

1036

960

1037

961

class TestIsEnabledCmd(TestCmd):

1038

962

def test_is_enabled(self):

1039

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

1040

properties)

1041

for client, properties

1042

in self.clients.items()))

963

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

964

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

1043

965

def test_is_enabled_run_exits_successfully(self):

1044

966

with self.assertRaises(SystemExit) as e:

1045

IsEnabledCmd().run(self.one_client)

967

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

1046

968

if e.exception.code is not None:

1047

969

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

1048

970

else:

1050

972

def test_is_enabled_run_exits_with_failure(self):

1051

973

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

1052

974

with self.assertRaises(SystemExit) as e:

1053

IsEnabledCmd().run(self.one_client)

975

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

1054

976

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

1055

977

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

1056

978

else:

1065

987

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

1066

988

mandos = MockMandos()

1067

989

super(TestRemoveCmd, self).setUp()

1068

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

990

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

1069

991

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

1070

for clientpath in self.clients:

1071

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

992

for client in self.clients:

993

self.assertIn(("RemoveClient",

994

(client.__dbus_object_path__,)),

1072

995

mandos.calls)

1073

996

1074

997

class TestApproveCmd(TestCmd):

1075

998

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

999

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

1000

for client in self.clients:

1001

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

1080

1002

client.calls)

1081

1003

1082

1004

class TestDenyCmd(TestCmd):

1083

1005

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

1006

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

1007

for client in self.clients:

1008

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

1088

1009

client.calls)

1089

1010

1090

1011

class TestEnableCmd(TestCmd):

1091

1012

def test_enable(self):

1092

for clientpath in self.clients:

1093

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

1013

for client in self.clients:

1094

1014

client.attributes["Enabled"] = False

1095

1015

1096

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

1016

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

1097

1017

1098

for clientpath in self.clients:

1099

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

1018

for client in self.clients:

1100

1019

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

1101

1020

1102

1021

class TestDisableCmd(TestCmd):

1103

1022

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)

1023

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

1024

1025

for client in self.clients:

1107

1026

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

1108

1027

1109

1028

class Unique(object):

1119

1038

self.values_to_set)

1120

1039

for value_to_set, value_to_get in zip(self.values_to_set,

1121

1040

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]

1041

for client in self.clients:

1042

old_value = client.attributes[self.property]

1125

1043

self.assertNotIsInstance(old_value, Unique)

1126

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

1044

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

1127

1045

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]

1046

for client in self.clients:

1047

value = client.attributes[self.property]

1131

1048

self.assertNotIsInstance(value, Unique)

1132

1049

self.assertEqual(value, value_to_get)

1133

1050

def run_command(self, value, clients):

1134

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

1051

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

1135

1052

1136

1053

class TestBumpTimeoutCmd(TestPropertyCmd):

1137

1054

command = BumpTimeoutCmd

1138

propname = "LastCheckedOK"

1055

property = "LastCheckedOK"

1139

1056

values_to_set = [""]

1140

1057

1141

1058

class TestStartCheckerCmd(TestPropertyCmd):

1142

1059

command = StartCheckerCmd

1143

propname = "CheckerRunning"

1060

property = "CheckerRunning"

1144

1061

values_to_set = [dbus.Boolean(True)]

1145

1062

1146

1063

class TestStopCheckerCmd(TestPropertyCmd):

1147

1064

command = StopCheckerCmd

1148

propname = "CheckerRunning"

1065

property = "CheckerRunning"

1149

1066

values_to_set = [dbus.Boolean(False)]

1150

1067

1151

1068

class TestApproveByDefaultCmd(TestPropertyCmd):

1152

1069

command = ApproveByDefaultCmd

1153

propname = "ApprovedByDefault"

1070

property = "ApprovedByDefault"

1154

1071

values_to_set = [dbus.Boolean(True)]

1155

1072

1156

1073

class TestDenyByDefaultCmd(TestPropertyCmd):

1157

1074

command = DenyByDefaultCmd

1158

propname = "ApprovedByDefault"

1075

property = "ApprovedByDefault"

1159

1076

values_to_set = [dbus.Boolean(False)]

1160

1077

1161

class TestPropertyValueCmd(TestPropertyCmd):

1162

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

1078

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1079

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

1080

ValueArgumentMixIn"""

1163

1081

def runTest(self):

1164

if type(self) is TestPropertyValueCmd:

1082

if type(self) is TestValueArgumentPropertyCmd:

1165

1083

return

1166

return super(TestPropertyValueCmd, self).runTest()

1084

return super(TestValueArgumentPropertyCmd, self).runTest()

1167

1085

def run_command(self, value, clients):

1168

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

1086

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

1169

1087

1170

class TestSetCheckerCmd(TestPropertyValueCmd):

1088

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1171

1089

command = SetCheckerCmd

1172

propname = "Checker"

1090

property = "Checker"

1173

1091

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

1174

1092

1175

class TestSetHostCmd(TestPropertyValueCmd):

1093

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1176

1094

command = SetHostCmd

1177

propname = "Host"

1095

property = "Host"

1178

1096

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

1179

1097

1180

class TestSetSecretCmd(TestPropertyValueCmd):

1098

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1181

1099

command = SetSecretCmd

1182

propname = "Secret"

1183

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

1100

property = "Secret"

1101

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

1184

1102

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

1185

1103

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

1186

1104

1187

class TestSetTimeoutCmd(TestPropertyValueCmd):

1105

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1188

1106

command = SetTimeoutCmd

1189

propname = "Timeout"

1107

property = "Timeout"

1190

1108

values_to_set = [datetime.timedelta(),

1191

1109

datetime.timedelta(minutes=5),

1192

1110

datetime.timedelta(seconds=1),

1194

1112

datetime.timedelta(weeks=52)]

1195

1113

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

1196

1114

1197

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1115

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1198

1116

command = SetExtendedTimeoutCmd

1199

propname = "ExtendedTimeout"

1117

property = "ExtendedTimeout"

1200

1118

values_to_set = [datetime.timedelta(),

1201

1119

datetime.timedelta(minutes=5),

1202

1120

datetime.timedelta(seconds=1),

1204

1122

datetime.timedelta(weeks=52)]

1205

1123

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

1206

1124

1207

class TestSetIntervalCmd(TestPropertyValueCmd):

1125

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1208

1126

command = SetIntervalCmd

1209

propname = "Interval"

1127

property = "Interval"

1210

1128

values_to_set = [datetime.timedelta(),

1211

1129

datetime.timedelta(minutes=5),

1212

1130

datetime.timedelta(seconds=1),

1214

1132

datetime.timedelta(weeks=52)]

1215

1133

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

1216

1134

1217

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1135

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1218

1136

command = SetApprovalDelayCmd

1219

propname = "ApprovalDelay"

1137

property = "ApprovalDelay"

1220

1138

values_to_set = [datetime.timedelta(),

1221

1139

datetime.timedelta(minutes=5),

1222

1140

datetime.timedelta(seconds=1),

1224

1142

datetime.timedelta(weeks=52)]

1225

1143

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

1226

1144

1227

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1145

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1228

1146

command = SetApprovalDurationCmd

1229

propname = "ApprovalDuration"

1147

property = "ApprovalDuration"

1230

1148

values_to_set = [datetime.timedelta(),

1231

1149

datetime.timedelta(minutes=5),

1232

1150

datetime.timedelta(seconds=1),

1238

1156

def setUp(self):

1239

1157

self.parser = argparse.ArgumentParser()

1240

1158

add_command_line_options(self.parser)

1241

def assert_command_from_args(self, args, command_cls,

1242

**cmd_attrs):

1159

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

1243

1160

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

1244

1161

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

1245

1162

options = self.parser.parse_args(args)

1413

1330

def test_is_enabled_short(self):

1414

1331

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

1415

1332

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

1333

1567

1334

1568

1335

def should_only_run_tests():