/mandos/trunk : revision 1054

64

65

locale.setlocale(locale.LC_ALL, "")

66

67

domain = "se.recompile"

68

busname = domain + ".Mandos"

69

server_path = "/"

70

server_interface = domain + ".Mandos"

71

client_interface = domain + ".Mandos.Client"

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

72

73

version = "1.8.3"

73

74

75

276

277

# Abstract classes first

277

278

class Command(object):

278

279

"""Abstract class for commands"""

279

def run(self, mandos, clients):

280

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

280

281

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

281

282

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

282

283

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

283

284

self.mandos = mandos

284

for client, properties in clients.items():

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)

285

289

self.run_on_one_client(client, properties)

286

290

287

291

class PrintCmd(Command):

293

297

"LastApprovalRequest", "ApprovalDelay",

294

298

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

295

299

"Expires", "LastCheckerStatus")

296

def run(self, mandos, clients):

297

print(self.output(clients))

300

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

301

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

302

def output(self, clients):

303

raise NotImplementedError()

298

304

299

305

class PropertyCmd(Command):

300

306

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

301

307

def run_on_one_client(self, client, properties):

302

308

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

303

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

309

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

304

310

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

311

dbus.PROPERTIES_IFACE, client_dbus_interface,

312

self.propname, self.value_to_set

307

313

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

308

314

else bool(self.value_to_set))

309

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

315

client.Set(client_dbus_interface, self.propname,

316

self.value_to_set,

310

317

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

311

321

312

class ValueArgumentMixIn(object):

313

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

322

class PropertyValueCmd(PropertyCmd):

323

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

314

324

def __init__(self, value):

315

325

self.value_to_set = value

316

326

317

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

318

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

319

milliseconds."""

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

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

329

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

320

330

@property

321

331

def value_to_set(self):

322

332

return self._vts

323

333

@value_to_set.setter

324

334

def value_to_set(self, value):

325

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

335

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

326

336

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

327

337

328

338

# Actual (non-abstract) command classes

332

342

self.verbose = verbose

333

343

334

344

def output(self, clients):

335

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

345

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

346

"LastCheckedOK")

336

347

keywords = default_keywords

337

348

if self.verbose:

338

349

keywords = self.all_keywords

339

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

350

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

340

351

341

352

class TableOfClients(object):

342

353

tableheaders = {

428

439

return value

429

440

430

441

class IsEnabledCmd(Command):

431

def run_on_one_client(self, client, properties):

442

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

443

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

432

444

if self.is_enabled(client, properties):

433

445

sys.exit(0)

434

446

sys.exit(1)

435

447

def is_enabled(self, client, properties):

436

return bool(properties["Enabled"])

448

return properties["Enabled"]

437

449

438

450

class RemoveCmd(Command):

439

451

def run_on_one_client(self, client, properties):

440

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

441

server_path, server_interface,

452

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

453

server_dbus_path, server_dbus_interface,

442

454

str(client.__dbus_object_path__))

443

455

self.mandos.RemoveClient(client.__dbus_object_path__)

444

456

445

457

class ApproveCmd(Command):

446

458

def run_on_one_client(self, client, properties):

447

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

448

client.__dbus_object_path__, client_interface)

459

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

460

client.__dbus_object_path__, client_dbus_interface)

449

461

client.Approve(dbus.Boolean(True),

450

dbus_interface=client_interface)

462

dbus_interface=client_dbus_interface)

451

463

452

464

class DenyCmd(Command):

453

465

def run_on_one_client(self, client, properties):

454

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

455

client.__dbus_object_path__, client_interface)

466

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

467

client.__dbus_object_path__, client_dbus_interface)

456

468

client.Approve(dbus.Boolean(False),

457

dbus_interface=client_interface)

469

dbus_interface=client_dbus_interface)

458

470

459

471

class EnableCmd(PropertyCmd):

460

property = "Enabled"

472

propname = "Enabled"

461

473

value_to_set = dbus.Boolean(True)

462

474

463

475

class DisableCmd(PropertyCmd):

464

property = "Enabled"

476

propname = "Enabled"

465

477

value_to_set = dbus.Boolean(False)

466

478

467

479

class BumpTimeoutCmd(PropertyCmd):

468

property = "LastCheckedOK"

480

propname = "LastCheckedOK"

469

481

value_to_set = ""

470

482

471

483

class StartCheckerCmd(PropertyCmd):

472

property = "CheckerRunning"

484

propname = "CheckerRunning"

473

485

value_to_set = dbus.Boolean(True)

474

486

475

487

class StopCheckerCmd(PropertyCmd):

476

property = "CheckerRunning"

488

propname = "CheckerRunning"

477

489

value_to_set = dbus.Boolean(False)

478

490

479

491

class ApproveByDefaultCmd(PropertyCmd):

480

property = "ApprovedByDefault"

492

propname = "ApprovedByDefault"

481

493

value_to_set = dbus.Boolean(True)

482

494

483

495

class DenyByDefaultCmd(PropertyCmd):

484

property = "ApprovedByDefault"

496

propname = "ApprovedByDefault"

485

497

value_to_set = dbus.Boolean(False)

486

498

487

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

488

property = "Checker"

489

490

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

491

property = "Host"

492

493

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

494

507

@property

495

508

def value_to_set(self):

496

509

return self._vts

499

512

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

500

513

self._vts = value.read()

501

514

value.close()

502

property = "Secret"

503

504

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

505

property = "Timeout"

506

507

class SetExtendedTimeoutCmd(PropertyCmd,

508

MillisecondsValueArgumentMixIn):

509

property = "ExtendedTimeout"

510

511

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

512

property = "Interval"

513

514

class SetApprovalDelayCmd(PropertyCmd,

515

MillisecondsValueArgumentMixIn):

516

property = "ApprovalDelay"

517

518

class SetApprovalDurationCmd(PropertyCmd,

519

MillisecondsValueArgumentMixIn):

520

property = "ApprovalDuration"

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"

521

530

522

531

def add_command_line_options(parser):

523

532

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

611

620

if options.is_enabled:

612

621

commands.append(IsEnabledCmd())

613

622

614

if options.remove:

615

commands.append(RemoveCmd())

616

617

623

if options.checker is not None:

618

624

commands.append(SetCheckerCmd(options.checker))

619

625

652

658

if options.deny:

653

659

commands.append(DenyCmd())

654

660

661

if options.remove:

662

commands.append(RemoveCmd())

663

655

664

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

656

665

# optionally verbosely

657

666

if not commands:

695

704

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

696

705

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

697

706

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

698

712

699

713

700

714

def main():

713

727

714

728

try:

715

729

bus = dbus.SystemBus()

716

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

717

server_path)

718

mandos_dbus_objc = bus.get_object(busname, server_path)

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)

719

734

except dbus.exceptions.DBusException:

720

735

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

721

736

sys.exit(1)

722

737

723

738

mandos_serv = dbus.Interface(mandos_dbus_objc,

724

dbus_interface=server_interface)

739

dbus_interface=server_dbus_interface)

725

740

mandos_serv_object_manager = dbus.Interface(

726

741

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

727

742

733

748

dbus_filter = NullFilter()

734

749

try:

735

750

dbus_logger.addFilter(dbus_filter)

736

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

737

server_path, dbus.OBJECT_MANAGER_IFACE)

738

mandos_clients = {path: ifs_and_props[client_interface]

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]

739

754

for path, ifs_and_props in

740

755

mandos_serv_object_manager

741

756

.GetManagedObjects().items()

742

if client_interface in ifs_and_props}

757

if client_dbus_interface in ifs_and_props}

743

758

except dbus.exceptions.DBusException as e:

744

759

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

745

760

"\n%s", e)

752

767

clients = {}

753

768

754

769

if not clientnames:

755

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

756

for path, properties in mandos_clients.items()}

770

clients = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

757

772

else:

758

773

for name in clientnames:

759

for path, client in mandos_clients.items():

760

if client["Name"] == name:

761

client_objc = bus.get_object(busname, path)

762

clients[client_objc] = client

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

763

777

break

764

778

else:

765

779

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

768

782

# Run all commands on clients

769

783

commands = commands_from_options(options)

770

784

for command in commands:

771

command.run(mandos_serv, clients)

785

command.run(clients, bus, mandos_serv)

772

786

773

787

774

788

class Test_milliseconds_to_string(unittest.TestCase):

823

837

testcase = self

824

838

class MockClient(object):

825

839

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

826

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

840

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

827

841

self.attributes = attributes

828

842

self.attributes["Name"] = name

829

843

self.calls = []

830

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

831

testcase.assertEqual(interface, client_interface)

832

testcase.assertEqual(dbus_interface,

833

dbus.PROPERTIES_IFACE)

834

self.attributes[property] = value

835

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

836

testcase.assertEqual(interface, client_interface)

837

testcase.assertEqual(dbus_interface,

838

dbus.PROPERTIES_IFACE)

839

return self.attributes[property]

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]

840

854

def Approve(self, approve, dbus_interface):

841

testcase.assertEqual(dbus_interface, client_interface)

855

testcase.assertEqual(dbus_interface,

856

client_dbus_interface)

842

857

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

843

858

dbus_interface)))

844

859

self.client = MockClient(

891

906

LastCheckerStatus=-2)

892

907

self.clients = collections.OrderedDict(

893

908

[

894

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

895

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

909

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

910

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

896

911

])

897

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

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

898

924

899

925

class TestPrintTableCmd(TestCmd):

900

926

def test_normal(self):

901

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

902

expected_output = """

903

Name Enabled Timeout Last Successful Check

904

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

905

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

906

"""[1:-1]

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

))

907

933

self.assertEqual(output, expected_output)

908

934

def test_verbose(self):

909

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

910

expected_output = """

911

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

912

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

913

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

914

"""[1:-1]

935

output = PrintTableCmd(verbose=True).output(

936

self.clients.values())

937

columns = (

938

(

939

"Name ",

940

"foo ",

941

"barbar ",

942

),(

943

"Enabled ",

944

"Yes ",

945

"Yes ",

946

),(

947

"Timeout ",

948

"00:05:00 ",

949

"00:05:00 ",

950

),(

951

"Last Successful Check ",

952

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

953

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

954

),(

955

"Created ",

956

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

957

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

958

),(

959

"Interval ",

960

"00:02:00 ",

961

"00:02:00 ",

962

),(

963

"Host ",

964

"foo.example.org ",

965

"192.0.2.3 ",

966

),(

967

("Key ID "

968

" "),

969

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

970

"ea90a27cddb12 "),

971

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

972

"46c2b020f441c "),

973

),(

974

"Fingerprint ",

975

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

976

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

977

),(

978

"Check Is Running ",

979

"No ",

980

"Yes ",

981

),(

982

"Last Enabled ",

983

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

984

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

985

),(

986

"Approval Is Pending ",

987

"No ",

988

"No ",

989

),(

990

"Approved By Default ",

991

"Yes ",

992

"No ",

993

),(

994

"Last Approval Request ",

995

" ",

996

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

997

),(

998

"Approval Delay ",

999

"00:00:00 ",

1000

"00:00:30 ",

1001

),(

1002

"Approval Duration ",

1003

"00:00:01 ",

1004

"00:00:01 ",

1005

),(

1006

"Checker ",

1007

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

1008

": ",

1009

),(

1010

"Extended Timeout ",

1011

"00:15:00 ",

1012

"00:15:00 ",

1013

),(

1014

"Expires ",

1015

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

1016

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

1017

),(

1018

"Last Checker Status",

1019

"0 ",

1020

"-2 ",

1021

)

1022

)

1023

num_lines = max(len(rows) for rows in columns)

1024

expected_output = "\n".join("".join(rows[line]

1025

for rows in columns)

1026

for line in range(num_lines))

915

1027

self.assertEqual(output, expected_output)

916

1028

def test_one_client(self):

917

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

1029

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

918

1030

expected_output = """

919

1031

Name Enabled Timeout Last Successful Check

920

1032

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

985

1097

986

1098

class TestIsEnabledCmd(TestCmd):

987

1099

def test_is_enabled(self):

988

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

989

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

1100

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

1101

properties)

1102

for client, properties

1103

in self.clients.items()))

990

1104

def test_is_enabled_run_exits_successfully(self):

991

1105

with self.assertRaises(SystemExit) as e:

992

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

1106

IsEnabledCmd().run(self.one_client)

993

1107

if e.exception.code is not None:

994

1108

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

995

1109

else:

997

1111

def test_is_enabled_run_exits_with_failure(self):

998

1112

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

999

1113

with self.assertRaises(SystemExit) as e:

1000

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

1114

IsEnabledCmd().run(self.one_client)

1001

1115

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

1002

1116

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

1003

1117

else:

1012

1126

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

1013

1127

mandos = MockMandos()

1014

1128

super(TestRemoveCmd, self).setUp()

1015

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

1129

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

1016

1130

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

1017

for client in self.clients:

1018

self.assertIn(("RemoveClient",

1019

(client.__dbus_object_path__,)),

1131

for clientpath in self.clients:

1132

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

1020

1133

mandos.calls)

1021

1134

1022

1135

class TestApproveCmd(TestCmd):

1023

1136

def test_approve(self):

1024

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

1025

for client in self.clients:

1026

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

1137

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

1138

for clientpath in self.clients:

1139

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

1140

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

1027

1141

client.calls)

1028

1142

1029

1143

class TestDenyCmd(TestCmd):

1030

1144

def test_deny(self):

1031

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

1032

for client in self.clients:

1033

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

1145

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

1146

for clientpath in self.clients:

1147

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

1148

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

1034

1149

client.calls)

1035

1150

1036

1151

class TestEnableCmd(TestCmd):

1037

1152

def test_enable(self):

1038

for client in self.clients:

1153

for clientpath in self.clients:

1154

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

1039

1155

client.attributes["Enabled"] = False

1040

1156

1041

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

1157

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

1042

1158

1043

for client in self.clients:

1159

for clientpath in self.clients:

1160

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

1044

1161

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

1045

1162

1046

1163

class TestDisableCmd(TestCmd):

1047

1164

def test_disable(self):

1048

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

1049

1050

for client in self.clients:

1165

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

1166

for clientpath in self.clients:

1167

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

1051

1168

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

1052

1169

1053

1170

class Unique(object):

1063

1180

self.values_to_set)

1064

1181

for value_to_set, value_to_get in zip(self.values_to_set,

1065

1182

values_to_get):

1066

for client in self.clients:

1067

old_value = client.attributes[self.property]

1183

for clientpath in self.clients:

1184

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

1185

old_value = client.attributes[self.propname]

1068

1186

self.assertNotIsInstance(old_value, Unique)

1069

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

1187

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

1070

1188

self.run_command(value_to_set, self.clients)

1071

for client in self.clients:

1072

value = client.attributes[self.property]

1189

for clientpath in self.clients:

1190

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

1191

value = client.attributes[self.propname]

1073

1192

self.assertNotIsInstance(value, Unique)

1074

1193

self.assertEqual(value, value_to_get)

1075

1194

def run_command(self, value, clients):

1076

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

1195

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

1077

1196

1078

1197

class TestBumpTimeoutCmd(TestPropertyCmd):

1079

1198

command = BumpTimeoutCmd

1080

property = "LastCheckedOK"

1199

propname = "LastCheckedOK"

1081

1200

values_to_set = [""]

1082

1201

1083

1202

class TestStartCheckerCmd(TestPropertyCmd):

1084

1203

command = StartCheckerCmd

1085

property = "CheckerRunning"

1204

propname = "CheckerRunning"

1086

1205

values_to_set = [dbus.Boolean(True)]

1087

1206

1088

1207

class TestStopCheckerCmd(TestPropertyCmd):

1089

1208

command = StopCheckerCmd

1090

property = "CheckerRunning"

1209

propname = "CheckerRunning"

1091

1210

values_to_set = [dbus.Boolean(False)]

1092

1211

1093

1212

class TestApproveByDefaultCmd(TestPropertyCmd):

1094

1213

command = ApproveByDefaultCmd

1095

property = "ApprovedByDefault"

1214

propname = "ApprovedByDefault"

1096

1215

values_to_set = [dbus.Boolean(True)]

1097

1216

1098

1217

class TestDenyByDefaultCmd(TestPropertyCmd):

1099

1218

command = DenyByDefaultCmd

1100

property = "ApprovedByDefault"

1219

propname = "ApprovedByDefault"

1101

1220

values_to_set = [dbus.Boolean(False)]

1102

1221

1103

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1104

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

1105

ValueArgumentMixIn"""

1222

class TestPropertyValueCmd(TestPropertyCmd):

1223

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

1106

1224

def runTest(self):

1107

if type(self) is TestValueArgumentPropertyCmd:

1225

if type(self) is TestPropertyValueCmd:

1108

1226

return

1109

return super(TestValueArgumentPropertyCmd, self).runTest()

1227

return super(TestPropertyValueCmd, self).runTest()

1110

1228

def run_command(self, value, clients):

1111

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

1229

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

1112

1230

1113

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1231

class TestSetCheckerCmd(TestPropertyValueCmd):

1114

1232

command = SetCheckerCmd

1115

property = "Checker"

1233

propname = "Checker"

1116

1234

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

1117

1235

1118

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1236

class TestSetHostCmd(TestPropertyValueCmd):

1119

1237

command = SetHostCmd

1120

property = "Host"

1238

propname = "Host"

1121

1239

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

1122

1240

1123

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1241

class TestSetSecretCmd(TestPropertyValueCmd):

1124

1242

command = SetSecretCmd

1125

property = "Secret"

1243

propname = "Secret"

1126

1244

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

1127

1245

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

1128

1246

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

1129

1247

1130

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1248

class TestSetTimeoutCmd(TestPropertyValueCmd):

1131

1249

command = SetTimeoutCmd

1132

property = "Timeout"

1250

propname = "Timeout"

1133

1251

values_to_set = [datetime.timedelta(),

1134

1252

datetime.timedelta(minutes=5),

1135

1253

datetime.timedelta(seconds=1),

1137

1255

datetime.timedelta(weeks=52)]

1138

1256

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

1139

1257

1140

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1258

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1141

1259

command = SetExtendedTimeoutCmd

1142

property = "ExtendedTimeout"

1260

propname = "ExtendedTimeout"

1143

1261

values_to_set = [datetime.timedelta(),

1144

1262

datetime.timedelta(minutes=5),

1145

1263

datetime.timedelta(seconds=1),

1147

1265

datetime.timedelta(weeks=52)]

1148

1266

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

1149

1267

1150

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1268

class TestSetIntervalCmd(TestPropertyValueCmd):

1151

1269

command = SetIntervalCmd

1152

property = "Interval"

1270

propname = "Interval"

1153

1271

values_to_set = [datetime.timedelta(),

1154

1272

datetime.timedelta(minutes=5),

1155

1273

datetime.timedelta(seconds=1),

1157

1275

datetime.timedelta(weeks=52)]

1158

1276

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

1159

1277

1160

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1278

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1161

1279

command = SetApprovalDelayCmd

1162

property = "ApprovalDelay"

1280

propname = "ApprovalDelay"

1163

1281

values_to_set = [datetime.timedelta(),

1164

1282

datetime.timedelta(minutes=5),

1165

1283

datetime.timedelta(seconds=1),

1167

1285

datetime.timedelta(weeks=52)]

1168

1286

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

1169

1287

1170

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1288

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1171

1289

command = SetApprovalDurationCmd

1172

property = "ApprovalDuration"

1290

propname = "ApprovalDuration"

1173

1291

values_to_set = [datetime.timedelta(),

1174

1292

datetime.timedelta(minutes=5),

1175

1293

datetime.timedelta(seconds=1),

1181

1299

def setUp(self):

1182

1300

self.parser = argparse.ArgumentParser()

1183

1301

add_command_line_options(self.parser)

1184

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

1302

def assert_command_from_args(self, args, command_cls,

1303

**cmd_attrs):

1185

1304

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

1186

1305

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

1187

1306

options = self.parser.parse_args(args)

1355

1474

def test_is_enabled_short(self):

1356

1475

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

1357

1476

1477

def test_deny_before_remove(self):

1478

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

1479

"foo"])

1480

check_option_syntax(self.parser, options)

1481

commands = commands_from_options(options)

1482

self.assertEqual(len(commands), 2)

1483

self.assertIsInstance(commands[0], DenyCmd)

1484

self.assertIsInstance(commands[1], RemoveCmd)

1485

1486

def test_deny_before_remove_reversed(self):

1487

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

1488

"--all"])

1489

check_option_syntax(self.parser, options)

1490

commands = commands_from_options(options)

1491

self.assertEqual(len(commands), 2)

1492

self.assertIsInstance(commands[0], DenyCmd)

1493

self.assertIsInstance(commands[1], RemoveCmd)

1494

1358

1495

1359

1496

class Test_check_option_syntax(unittest.TestCase):

1360

1497

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

1393

1530

with self.temporarily_suppress_stderr():

1394

1531

yield

1395

1532

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

1396

# https://docs.python.org/3/library/argparse.html#exiting-methods

1533

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

1534

# /argparse.html#exiting-methods

1397

1535

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

1398

1536

1399

1537

@staticmethod

1475

1613

with self.assertParseError():

1476

1614

self.check_option_syntax(options)

1477

1615

1616

def test_remove_can_only_be_combined_with_action_deny(self):

1617

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

1618

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

1619

continue

1620

options = self.parser.parse_args()

1621

setattr(options, action, value)

1622

options.all = True

1623

options.remove = True

1624

with self.assertParseError():

1625

self.check_option_syntax(options)

1626

1478

1627

1479

1628

1480

1629

def should_only_run_tests():