/mandos/trunk : revision 1080

61

62

if sys.version_info.major == 2:

63

str = unicode

64

import StringIO

65

io.StringIO = StringIO.StringIO

66

64

67

65

locale.setlocale(locale.LC_ALL, "")

68

66

83

81

84

82

def main():

85

83

parser = argparse.ArgumentParser()

84

86

85

add_command_line_options(parser)

87

86

88

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

463

464

465

class NullFilter(logging.Filter):

465

466

def filter(self, record):

612

613

"Checker", "ExtendedTimeout", "Expires",

613

614

"LastCheckerStatus")

614

615

616

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

617

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

618

619

def output(self, clients):

620

raise NotImplementedError()

621

615

622

616

623

class DumpJSON(Output):

617

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

624

def output(self, clients):

618

625

data = {client["Name"]:

619

626

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

620

627

for key in self.all_keywords}

621

for client in clients.values()}

622

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

628

for client in clients}

629

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

623

630

624

631

@staticmethod

625

632

def dbus_boolean_to_bool(value):

632

639

def __init__(self, verbose=False):

633

640

self.verbose = verbose

634

641

635

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

642

def output(self, clients):

636

643

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

637

644

"LastCheckedOK")

638

645

keywords = default_keywords

639

646

if self.verbose:

640

647

keywords = self.all_keywords

641

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

648

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

642

649

643

650

class TableOfClients(object):

644

651

tableheaders = {

725

732

seconds=td.seconds % 60))

726

733

727

734

728

class PropertySetter(Base):

735

class Property(Base):

729

736

"Abstract class for Actions for setting one client property"

730

737

731

738

def run_on_one_client(self, client, properties):

746

753

raise NotImplementedError()

747

754

748

755

749

class Enable(PropertySetter):

756

class Enable(Property):

750

757

propname = "Enabled"

751

758

value_to_set = dbus.Boolean(True)

752

759

753

760

754

class Disable(PropertySetter):

761

class Disable(Property):

755

762

propname = "Enabled"

756

763

value_to_set = dbus.Boolean(False)

757

764

758

765

759

class BumpTimeout(PropertySetter):

766

class BumpTimeout(Property):

760

767

propname = "LastCheckedOK"

761

768

value_to_set = ""

762

769

763

770

764

class StartChecker(PropertySetter):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(True)

767

768

769

class StopChecker(PropertySetter):

770

propname = "CheckerRunning"

771

value_to_set = dbus.Boolean(False)

772

773

774

class ApproveByDefault(PropertySetter):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(True)

777

778

779

class DenyByDefault(PropertySetter):

780

propname = "ApprovedByDefault"

781

value_to_set = dbus.Boolean(False)

782

783

784

class PropertySetterValue(PropertySetter):

785

"""Abstract class for PropertySetter recieving a value as

786

constructor argument instead of a class attribute."""

771

class StartChecker(Property):

772

propname = "CheckerRunning"

773

value_to_set = dbus.Boolean(True)

774

775

776

class StopChecker(Property):

777

propname = "CheckerRunning"

778

value_to_set = dbus.Boolean(False)

779

780

781

class ApproveByDefault(Property):

782

propname = "ApprovedByDefault"

783

value_to_set = dbus.Boolean(True)

784

785

786

class DenyByDefault(Property):

787

propname = "ApprovedByDefault"

788

value_to_set = dbus.Boolean(False)

789

790

791

class PropertyValue(Property):

792

"Abstract class for Property recieving a value as argument"

787

793

def __init__(self, value):

788

794

self.value_to_set = value

789

795

790

796

791

class SetChecker(PropertySetterValue):

797

class SetChecker(PropertyValue):

792

798

propname = "Checker"

793

799

794

800

795

class SetHost(PropertySetterValue):

801

class SetHost(PropertyValue):

796

802

propname = "Host"

797

803

798

804

799

class SetSecret(PropertySetterValue):

805

class SetSecret(PropertyValue):

800

806

propname = "Secret"

801

807

802

808

@property

810

816

value.close()

811

817

812

818

813

class PropertySetterValueMilliseconds(PropertySetterValue):

814

"""Abstract class for PropertySetterValue taking a value

815

argument as a datetime.timedelta() but should store it as

816

milliseconds."""

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

817

822

818

823

@property

819

824

def value_to_set(self):

825

830

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

826

831

827

832

828

class SetTimeout(PropertySetterValueMilliseconds):

833

class SetTimeout(MillisecondsPropertyValueArgument):

829

834

propname = "Timeout"

830

835

831

836

832

class SetExtendedTimeout(PropertySetterValueMilliseconds):

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

833

838

propname = "ExtendedTimeout"

834

839

835

840

836

class SetInterval(PropertySetterValueMilliseconds):

841

class SetInterval(MillisecondsPropertyValueArgument):

837

842

propname = "Interval"

838

843

839

844

840

class SetApprovalDelay(PropertySetterValueMilliseconds):

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

841

846

propname = "ApprovalDelay"

842

847

843

848

844

class SetApprovalDuration(PropertySetterValueMilliseconds):

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

845

850

propname = "ApprovalDuration"

846

851

847

852

886

891

# tests, which is run by doctest.

887

892

888

893

def test_rfc3339_zero_seconds(self):

889

self.assertEqual(datetime.timedelta(),

890

string_to_delta("PT0S"))

894

self.assertEqual(string_to_delta("PT0S"),

895

datetime.timedelta())

891

896

892

897

def test_rfc3339_zero_days(self):

893

self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))

898

self.assertEqual(string_to_delta("P0D"),

899

datetime.timedelta())

894

900

895

901

def test_rfc3339_one_second(self):

896

self.assertEqual(datetime.timedelta(0, 1),

897

string_to_delta("PT1S"))

902

self.assertEqual(string_to_delta("PT1S"),

903

datetime.timedelta(0, 1))

898

904

899

905

def test_rfc3339_two_hours(self):

900

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

901

string_to_delta("PT2H"))

906

self.assertEqual(string_to_delta("PT2H"),

907

datetime.timedelta(0, 7200))

902

908

903

909

def test_falls_back_to_pre_1_6_1_with_warning(self):

904

910

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

905

911

value = string_to_delta("2h")

906

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

912

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

907

913

908

914

909

915

class Test_check_option_syntax(unittest.TestCase):

952

958

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

953

959

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

954

960

# /argparse.html#exiting-methods

955

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

961

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

956

962

957

963

@staticmethod

958

964

@contextlib.contextmanager

959

965

def redirect_stderr_to_devnull():

960

old_stderr = sys.stderr

961

with contextlib.closing(open(os.devnull, "w")) as null:

962

sys.stderr = null

963

try:

964

yield

965

finally:

966

sys.stderr = old_stderr

966

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

967

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

968

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

969

os.close(null)

970

try:

971

yield

972

finally:

973

# restore stderr

974

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

975

os.close(stderrcopy)

967

976

968

977

def check_option_syntax(self, options):

969

978

check_option_syntax(self.parser, options)

982

991

options = self.parser.parse_args()

983

992

setattr(options, action, value)

984

993

options.verbose = True

985

options.client = ["client"]

994

options.client = ["foo"]

986

995

with self.assertParseError():

987

996

self.check_option_syntax(options)

988

997

1018

1027

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

1019

1028

options = self.parser.parse_args()

1020

1029

setattr(options, action, value)

1021

options.client = ["client"]

1030

options.client = ["foo"]

1022

1031

self.check_option_syntax(options)

1023

1032

1024

1033

def test_one_client_with_all_actions_except_is_enabled(self):

1027

1036

if action == "is_enabled":

1028

1037

continue

1029

1038

setattr(options, action, value)

1030

options.client = ["client"]

1039

options.client = ["foo"]

1031

1040

self.check_option_syntax(options)

1032

1041

1033

1042

def test_two_clients_with_all_actions_except_is_enabled(self):

1036

1045

if action == "is_enabled":

1037

1046

continue

1038

1047

setattr(options, action, value)

1039

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

1048

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

1040

1049

self.check_option_syntax(options)

1041

1050

1042

1051

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1045

1054

continue

1046

1055

options = self.parser.parse_args()

1047

1056

setattr(options, action, value)

1048

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

1057

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

1049

1058

self.check_option_syntax(options)

1050

1059

1051

1060

def test_is_enabled_fails_without_client(self):

1057

1066

def test_is_enabled_fails_with_two_clients(self):

1058

1067

options = self.parser.parse_args()

1059

1068

options.is_enabled = True

1060

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

1069

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

1061

1070

with self.assertParseError():

1062

1071

self.check_option_syntax(options)

1063

1072

1080

1089

def get_object(mockbus_self, busname, dbus_path):

1081

1090

# Note that "self" is still the testcase instance,

1082

1091

# this MockBus instance is in "mockbus_self".

1083

self.assertEqual(dbus_busname, busname)

1084

self.assertEqual(server_dbus_path, dbus_path)

1092

self.assertEqual(busname, dbus_busname)

1093

self.assertEqual(dbus_path, server_dbus_path)

1085

1094

mockbus_self.called = True

1086

1095

return mockbus_self

1087

1096

1090

1099

self.assertTrue(mockbus.called)

1091

1100

1092

1101

def test_logs_and_exits_on_dbus_error(self):

1093

class FailingBusStub(object):

1102

class MockBusFailing(object):

1094

1103

def get_object(self, busname, dbus_path):

1095

1104

raise dbus.exceptions.DBusException("Test")

1096

1105

1097

1106

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

1098

1107

with self.assertRaises(SystemExit) as e:

1099

bus = get_mandos_dbus_object(bus=FailingBusStub())

1108

bus = get_mandos_dbus_object(bus=MockBusFailing())

1100

1109

1101

1110

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

1102

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

1111

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

1103

1112

else:

1104

1113

self.assertIsNotNone(e.exception.code)

1105

1114

1106

1115

1107

1116

class Test_get_managed_objects(TestCaseWithAssertLogs):

1108

1117

def test_calls_and_returns_GetManagedObjects(self):

1109

managed_objects = {"/clients/client": { "Name": "client"}}

1110

class ObjectManagerStub(object):

1118

managed_objects = {"/clients/foo": { "Name": "foo"}}

1119

class MockObjectManager(object):

1111

1120

def GetManagedObjects(self):

1112

1121

return managed_objects

1113

retval = get_managed_objects(ObjectManagerStub())

1122

retval = get_managed_objects(MockObjectManager())

1114

1123

self.assertDictEqual(managed_objects, retval)

1115

1124

1116

1125

def test_logs_and_exits_on_dbus_error(self):

1117

1126

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

1118

1127

1119

class ObjectManagerFailingStub(object):

1128

class MockObjectManagerFailing(object):

1120

1129

def GetManagedObjects(self):

1121

1130

dbus_logger.error("Test")

1122

1131

raise dbus.exceptions.DBusException("Test")

1133

1142

try:

1134

1143

with self.assertLogs(log, logging.CRITICAL) as watcher:

1135

1144

with self.assertRaises(SystemExit) as e:

1136

get_managed_objects(ObjectManagerFailingStub())

1145

get_managed_objects(MockObjectManagerFailing())

1137

1146

finally:

1138

1147

dbus_logger.removeFilter(counting_handler)

1139

1148

1140

1149

# Make sure the dbus logger was suppressed

1141

self.assertEqual(0, counting_handler.count)

1150

self.assertEqual(counting_handler.count, 0)

1142

1151

1143

1152

# Test that the dbus_logger still works

1144

1153

with self.assertLogs(dbus_logger, logging.ERROR):

1145

1154

dbus_logger.error("Test")

1146

1155

1147

1156

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

1148

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

1157

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

1149

1158

else:

1150

1159

self.assertIsNotNone(e.exception.code)

1151

1160

1156

1165

add_command_line_options(self.parser)

1157

1166

1158

1167

def test_is_enabled(self):

1159

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

1168

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

1160

1169

command.IsEnabled)

1161

1170

1162

1171

def assert_command_from_args(self, args, command_cls,

1166

1175

options = self.parser.parse_args(args)

1167

1176

check_option_syntax(self.parser, options)

1168

1177

commands = commands_from_options(options)

1169

self.assertEqual(1, len(commands))

1178

self.assertEqual(len(commands), 1)

1170

1179

command = commands[0]

1171

1180

self.assertIsInstance(command, command_cls)

1172

1181

for key, value in cmd_attrs.items():

1173

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

1182

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

1174

1183

1175

1184

def test_is_enabled_short(self):

1176

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

1185

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

1177

1186

command.IsEnabled)

1178

1187

1179

1188

def test_approve(self):

1180

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

1189

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

1181

1190

command.Approve)

1182

1191

1183

1192

def test_approve_short(self):

1184

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

1185

command.Approve)

1193

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

1186

1194

1187

1195

def test_deny(self):

1188

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

1189

command.Deny)

1196

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

1190

1197

1191

1198

def test_deny_short(self):

1192

self.assert_command_from_args(["-D", "client"], command.Deny)

1199

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

1193

1200

1194

1201

def test_remove(self):

1195

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

1202

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

1196

1203

command.Remove)

1197

1204

1198

1205

def test_deny_before_remove(self):

1199

1206

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

1200

"client"])

1207

"foo"])

1201

1208

check_option_syntax(self.parser, options)

1202

1209

commands = commands_from_options(options)

1203

self.assertEqual(2, len(commands))

1210

self.assertEqual(len(commands), 2)

1204

1211

self.assertIsInstance(commands[0], command.Deny)

1205

1212

self.assertIsInstance(commands[1], command.Remove)

1206

1213

1209

1216

"--all"])

1210

1217

check_option_syntax(self.parser, options)

1211

1218

commands = commands_from_options(options)

1212

self.assertEqual(2, len(commands))

1219

self.assertEqual(len(commands), 2)

1213

1220

self.assertIsInstance(commands[0], command.Deny)

1214

1221

self.assertIsInstance(commands[1], command.Remove)

1215

1222

1216

1223

def test_remove_short(self):

1217

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

1218

command.Remove)

1224

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

1219

1225

1220

1226

def test_dump_json(self):

1221

1227

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

1222

1228

command.DumpJSON)

1223

1229

1224

1230

def test_enable(self):

1225

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

1231

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

1226

1232

command.Enable)

1227

1233

1228

1234

def test_enable_short(self):

1229

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

1230

command.Enable)

1235

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

1231

1236

1232

1237

def test_disable(self):

1233

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

1238

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

1234

1239

command.Disable)

1235

1240

1236

1241

def test_disable_short(self):

1237

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

1238

command.Disable)

1242

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

1239

1243

1240

1244

def test_bump_timeout(self):

1241

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

1245

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

1242

1246

command.BumpTimeout)

1243

1247

1244

1248

def test_bump_timeout_short(self):

1245

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

1249

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

1246

1250

command.BumpTimeout)

1247

1251

1248

1252

def test_start_checker(self):

1249

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

1253

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

1250

1254

command.StartChecker)

1251

1255

1252

1256

def test_stop_checker(self):

1253

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

1257

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

1254

1258

command.StopChecker)

1255

1259

1256

1260

def test_approve_by_default(self):

1257

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

1258

"client"],

1261

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

1259

1262

command.ApproveByDefault)

1260

1263

1261

1264

def test_deny_by_default(self):

1262

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

1265

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

1263

1266

command.DenyByDefault)

1264

1267

1265

1268

def test_checker(self):

1266

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

1269

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

1267

1270

command.SetChecker,

1268

1271

value_to_set=":")

1269

1272

1270

1273

def test_checker_empty(self):

1271

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

1274

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

1272

1275

command.SetChecker,

1273

1276

value_to_set="")

1274

1277

1275

1278

def test_checker_short(self):

1276

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

1279

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

1277

1280

command.SetChecker,

1278

1281

value_to_set=":")

1279

1282

1280

1283

def test_host(self):

1281

self.assert_command_from_args(

1282

["--host", "client.example.org", "client"],

1283

command.SetHost, value_to_set="client.example.org")

1284

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

1285

"foo"], command.SetHost,

1286

value_to_set="foo.example.org")

1284

1287

1285

1288

def test_host_short(self):

1286

self.assert_command_from_args(

1287

["-H", "client.example.org", "client"], command.SetHost,

1288

value_to_set="client.example.org")

1289

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

1290

"foo"], command.SetHost,

1291

value_to_set="foo.example.org")

1289

1292

1290

1293

def test_secret_devnull(self):

1291

1294

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

1292

"client"], command.SetSecret,

1295

"foo"], command.SetSecret,

1293

1296

value_to_set=b"")

1294

1297

1295

1298

def test_secret_tempfile(self):

1298

1301

f.write(value)

1299

1302

f.seek(0)

1300

1303

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

1301

"client"],

1302

command.SetSecret,

1304

"foo"], command.SetSecret,

1303

1305

value_to_set=value)

1304

1306

1305

1307

def test_secret_devnull_short(self):

1306

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

1307

"client"], command.SetSecret,

1308

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

1309

command.SetSecret,

1308

1310

value_to_set=b"")

1309

1311

1310

1312

def test_secret_tempfile_short(self):

1312

1314

value = b"secret\0xyzzy\nbar"

1313

1315

f.write(value)

1314

1316

f.seek(0)

1315

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

1317

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

1316

1318

command.SetSecret,

1317

1319

value_to_set=value)

1318

1320

1319

1321

def test_timeout(self):

1320

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

1322

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

1321

1323

command.SetTimeout,

1322

1324

value_to_set=300000)

1323

1325

1324

1326

def test_timeout_short(self):

1325

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

1327

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

1326

1328

command.SetTimeout,

1327

1329

value_to_set=300000)

1328

1330

1329

1331

def test_extended_timeout(self):

1330

1332

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

1331

"client"],

1333

"foo"],

1332

1334

command.SetExtendedTimeout,

1333

1335

value_to_set=900000)

1334

1336

1335

1337

def test_interval(self):

1336

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

1337

"client"], command.SetInterval,

1338

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

1339

command.SetInterval,

1338

1340

value_to_set=120000)

1339

1341

1340

1342

def test_interval_short(self):

1341

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

1343

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

1342

1344

command.SetInterval,

1343

1345

value_to_set=120000)

1344

1346

1345

1347

def test_approval_delay(self):

1346

1348

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

1347

"client"],

1349

"foo"],

1348

1350

command.SetApprovalDelay,

1349

1351

value_to_set=30000)

1350

1352

1351

1353

def test_approval_duration(self):

1352

1354

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

1353

"client"],

1355

"foo"],

1354

1356

command.SetApprovalDuration,

1355

1357

value_to_set=1000)

1356

1358

1380

1382

self.attributes["Name"] = name

1381

1383

self.calls = []

1382

1384

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

1383

testcase.assertEqual(client_dbus_interface, interface)

1384

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1385

dbus_interface)

1385

testcase.assertEqual(interface, client_dbus_interface)

1386

testcase.assertEqual(dbus_interface,

1387

dbus.PROPERTIES_IFACE)

1386

1388

self.attributes[propname] = value

1387

1389

def Approve(self, approve, dbus_interface):

1388

testcase.assertEqual(client_dbus_interface,

1389

dbus_interface)

1390

testcase.assertEqual(dbus_interface,

1391

client_dbus_interface)

1390

1392

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

1391

1393

dbus_interface)))

1392

1394

self.client = MockClient(

1439

1441

LastCheckerStatus=-2)

1440

1442

self.clients = collections.OrderedDict(

1441

1443

[

1442

(self.client.__dbus_object_path__,

1443

self.client.attributes),

1444

(self.other_client.__dbus_object_path__,

1445

self.other_client.attributes),

1444

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

1445

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

1446

])

1447

self.one_client = {self.client.__dbus_object_path__:

1448

self.client.attributes}

1447

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

1449

1448

1450

1449

@property

1451

1450

def bus(self):

1452

class MockBus(object):

1451

class Bus(object):

1453

1452

@staticmethod

1454

1453

def get_object(client_bus_name, path):

1455

self.assertEqual(dbus_busname, client_bus_name)

1456

# Note: "self" here is the TestCmd instance, not the

1457

# MockBus instance, since this is a static method!

1458

if path == self.client.__dbus_object_path__:

1459

return self.client

1460

elif path == self.other_client.__dbus_object_path__:

1461

return self.other_client

1462

return MockBus()

1454

self.assertEqual(client_bus_name, dbus_busname)

1455

return {

1456

# Note: "self" here is the TestCmd instance, not

1457

# the Bus instance, since this is a static method!

1458

"/clients/foo": self.client,

1459

"/clients/barbar": self.other_client,

1460

}[path]

1461

return Bus()

1463

1462

1464

1463

1465

1464

class TestBaseCommands(TestCommand):

1466

1465

1467

def test_IsEnabled_exits_successfully(self):

1466

def test_IsEnabled(self):

1467

self.assertTrue(all(command.IsEnabled().is_enabled(client,

1468

properties)

1469

for client, properties

1470

in self.clients.items()))

1471

1472

def test_IsEnabled_run_exits_successfully(self):

1468

1473

with self.assertRaises(SystemExit) as e:

1469

1474

command.IsEnabled().run(self.one_client)

1470

1475

if e.exception.code is not None:

1471

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

1476

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

1472

1477

else:

1473

1478

self.assertIsNone(e.exception.code)

1474

1479

1475

def test_IsEnabled_exits_with_failure(self):

1480

def test_IsEnabled_run_exits_with_failure(self):

1476

1481

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

1477

1482

with self.assertRaises(SystemExit) as e:

1478

1483

command.IsEnabled().run(self.one_client)

1479

1484

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

1480

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

1485

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

1481

1486

else:

1482

1487

self.assertIsNotNone(e.exception.code)

1483

1488

1496

1501

client.calls)

1497

1502

1498

1503

def test_Remove(self):

1499

class MandosSpy(object):

1504

class MockMandos(object):

1500

1505

def __init__(self):

1501

1506

self.calls = []

1502

1507

def RemoveClient(self, dbus_path):

1503

1508

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

1504

mandos = MandosSpy()

1509

mandos = MockMandos()

1505

1510

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

1506

1511

for clientpath in self.clients:

1507

1512

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

1559

1564

}

1560

1565

1561

1566

def test_DumpJSON_normal(self):

1562

with self.capture_stdout_to_buffer() as buffer:

1563

command.DumpJSON().run(self.clients)

1564

json_data = json.loads(buffer.getvalue())

1565

self.assertDictEqual(self.expected_json, json_data)

1566

1567

@staticmethod

1568

@contextlib.contextmanager

1569

def capture_stdout_to_buffer():

1570

capture_buffer = io.StringIO()

1571

old_stdout = sys.stdout

1572

sys.stdout = capture_buffer

1573

try:

1574

yield capture_buffer

1575

finally:

1576

sys.stdout = old_stdout

1567

output = command.DumpJSON().output(self.clients.values())

1568

json_data = json.loads(output)

1569

self.assertDictEqual(json_data, self.expected_json)

1577

1570

1578

1571

def test_DumpJSON_one_client(self):

1579

with self.capture_stdout_to_buffer() as buffer:

1580

command.DumpJSON().run(self.one_client)

1581

json_data = json.loads(buffer.getvalue())

1572

output = command.DumpJSON().output(self.one_client.values())

1573

json_data = json.loads(output)

1582

1574

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

1583

self.assertDictEqual(expected_json, json_data)

1575

self.assertDictEqual(json_data, expected_json)

1584

1576

1585

1577

def test_PrintTable_normal(self):

1586

with self.capture_stdout_to_buffer() as buffer:

1587

command.PrintTable().run(self.clients)

1578

output = command.PrintTable().output(self.clients.values())

1588

1579

expected_output = "\n".join((

1589

1580

"Name Enabled Timeout Last Successful Check",

1590

1581

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

1591

1582

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

1592

)) + "\n"

1593

self.assertEqual(expected_output, buffer.getvalue())

1583

))

1584

self.assertEqual(output, expected_output)

1594

1585

1595

1586

def test_PrintTable_verbose(self):

1596

with self.capture_stdout_to_buffer() as buffer:

1597

command.PrintTable(verbose=True).run(self.clients)

1587

output = command.PrintTable(verbose=True).output(

1588

self.clients.values())

1598

1589

columns = (

1599

1590

(

1600

1591

"Name ",

1682

1673

)

1683

1674

)

1684

1675

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

1685

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

1686

for rows in columns)

1687

for line in range(num_lines))

1688

+ "\n")

1689

self.assertEqual(expected_output, buffer.getvalue())

1676

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

1677

for rows in columns)

1678

for line in range(num_lines))

1679

self.assertEqual(output, expected_output)

1690

1680

1691

1681

def test_PrintTable_one_client(self):

1692

with self.capture_stdout_to_buffer() as buffer:

1693

command.PrintTable().run(self.one_client)

1682

output = command.PrintTable().output(self.one_client.values())

1694

1683

expected_output = "\n".join((

1695

1684

"Name Enabled Timeout Last Successful Check",

1696

1685

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

1697

)) + "\n"

1698

self.assertEqual(expected_output, buffer.getvalue())

1699

1700

1701

class TestPropertySetterCmd(TestCommand):

1702

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

1686

))

1687

self.assertEqual(output, expected_output)

1688

1689

1690

class TestPropertyCmd(TestCommand):

1691

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

1703

1692

def runTest(self):

1704

1693

if not hasattr(self, "command"):

1705

1694

return

1716

1705

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

1717

1706

value = client.attributes[self.propname]

1718

1707

self.assertNotIsInstance(value, self.Unique)

1719

self.assertEqual(value_to_get, value)

1708

self.assertEqual(value, value_to_get)

1720

1709

1721

1710

class Unique(object):

1722

1711

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

1726

1715

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

1727

1716

1728

1717

1729

class TestEnableCmd(TestPropertySetterCmd):

1718

class TestEnableCmd(TestPropertyCmd):

1730

1719

command = command.Enable

1731

1720

propname = "Enabled"

1732

1721

values_to_set = [dbus.Boolean(True)]

1733

1722

1734

1723

1735

class TestDisableCmd(TestPropertySetterCmd):

1724

class TestDisableCmd(TestPropertyCmd):

1736

1725

command = command.Disable

1737

1726

propname = "Enabled"

1738

1727

values_to_set = [dbus.Boolean(False)]

1739

1728

1740

1729

1741

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1730

class TestBumpTimeoutCmd(TestPropertyCmd):

1742

1731

command = command.BumpTimeout

1743

1732

propname = "LastCheckedOK"

1744

1733

values_to_set = [""]

1745

1734

1746

1735

1747

class TestStartCheckerCmd(TestPropertySetterCmd):

1736

class TestStartCheckerCmd(TestPropertyCmd):

1748

1737

command = command.StartChecker

1749

1738

propname = "CheckerRunning"

1750

1739

values_to_set = [dbus.Boolean(True)]

1751

1740

1752

1741

1753

class TestStopCheckerCmd(TestPropertySetterCmd):

1742

class TestStopCheckerCmd(TestPropertyCmd):

1754

1743

command = command.StopChecker

1755

1744

propname = "CheckerRunning"

1756

1745

values_to_set = [dbus.Boolean(False)]

1757

1746

1758

1747

1759

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1748

class TestApproveByDefaultCmd(TestPropertyCmd):

1760

1749

command = command.ApproveByDefault

1761

1750

propname = "ApprovedByDefault"

1762

1751

values_to_set = [dbus.Boolean(True)]

1763

1752

1764

1753

1765

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1754

class TestDenyByDefaultCmd(TestPropertyCmd):

1766

1755

command = command.DenyByDefault

1767

1756

propname = "ApprovedByDefault"

1768

1757

values_to_set = [dbus.Boolean(False)]

1769

1758

1770

1759

1771

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1772

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

1760

class TestPropertyValueCmd(TestPropertyCmd):

1761

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

1773

1762

1774

1763

def runTest(self):

1775

if type(self) is TestPropertySetterValueCmd:

1764

if type(self) is TestPropertyValueCmd:

1776

1765

return

1777

return super(TestPropertySetterValueCmd, self).runTest()

1766

return super(TestPropertyValueCmd, self).runTest()

1778

1767

1779

1768

def run_command(self, value, clients):

1780

1769

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

1781

1770

1782

1771

1783

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1772

class TestSetCheckerCmd(TestPropertyValueCmd):

1784

1773

command = command.SetChecker

1785

1774

propname = "Checker"

1786

1775

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

1787

1776

1788

1777

1789

class TestSetHostCmd(TestPropertySetterValueCmd):

1778

class TestSetHostCmd(TestPropertyValueCmd):

1790

1779

command = command.SetHost

1791

1780

propname = "Host"

1792

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

1793

1794

1795

class TestSetSecretCmd(TestPropertySetterValueCmd):

1781

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

1782

1783

1784

class TestSetSecretCmd(TestPropertyValueCmd):

1796

1785

command = command.SetSecret

1797

1786

propname = "Secret"

1798

1787

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

1799

1788

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

1800

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

1801

1802

1803

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1789

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

1790

1791

1792

class TestSetTimeoutCmd(TestPropertyValueCmd):

1804

1793

command = command.SetTimeout

1805

1794

propname = "Timeout"

1806

1795

values_to_set = [datetime.timedelta(),

1808

1797

datetime.timedelta(seconds=1),

1809

1798

datetime.timedelta(weeks=1),

1810

1799

datetime.timedelta(weeks=52)]

1811

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

1812

1813

1814

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1800

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

1801

1802

1803

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1815

1804

command = command.SetExtendedTimeout

1816

1805

propname = "ExtendedTimeout"

1817

1806

values_to_set = [datetime.timedelta(),

1819

1808

datetime.timedelta(seconds=1),

1820

1809

datetime.timedelta(weeks=1),

1821

1810

datetime.timedelta(weeks=52)]

1822

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

1823

1824

1825

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1811

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

1812

1813

1814

class TestSetIntervalCmd(TestPropertyValueCmd):

1826

1815

command = command.SetInterval

1827

1816

propname = "Interval"

1828

1817

values_to_set = [datetime.timedelta(),

1830

1819

datetime.timedelta(seconds=1),

1831

1820

datetime.timedelta(weeks=1),

1832

1821

datetime.timedelta(weeks=52)]

1833

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

1834

1835

1836

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1822

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

1823

1824

1825

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1837

1826

command = command.SetApprovalDelay

1838

1827

propname = "ApprovalDelay"

1839

1828

values_to_set = [datetime.timedelta(),

1841

1830

datetime.timedelta(seconds=1),

1842

1831

datetime.timedelta(weeks=1),

1843

1832

datetime.timedelta(weeks=52)]

1844

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

1845

1846

1847

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1833

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

1834

1835

1836

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1848

1837

command = command.SetApprovalDuration

1849

1838

propname = "ApprovalDuration"

1850

1839

values_to_set = [datetime.timedelta(),

1852

1841

datetime.timedelta(seconds=1),

1853

1842

datetime.timedelta(weeks=1),

1854

1843

datetime.timedelta(weeks=52)]

1855

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

1844

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

1856

1845

1857

1846

1858

1847