/mandos/trunk : revision 1043

44

import logging

45

import io

46

import tempfile

47

import contextlib

47

48

49

import dbus

49

50

299

300

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

300

301

def run_on_one_client(self, client, properties):

301

302

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

303

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

304

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

307

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

308

else bool(self.value_to_set))

302

309

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

303

310

dbus_interface=dbus.PROPERTIES_IFACE)

304

311

316

323

@value_to_set.setter

317

324

def value_to_set(self, value):

318

325

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

319

self._vts = string_to_delta(value).total_seconds() * 1000

326

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

320

327

321

328

# Actual (non-abstract) command classes

322

329

430

437

431

438

class RemoveCmd(Command):

432

439

def run_on_one_client(self, client, properties):

440

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

441

server_path, server_interface,

442

str(client.__dbus_object_path__))

433

443

self.mandos.RemoveClient(client.__dbus_object_path__)

434

444

435

445

class ApproveCmd(Command):

436

446

def run_on_one_client(self, client, properties):

447

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

448

client.__dbus_object_path__, client_interface)

437

449

client.Approve(dbus.Boolean(True),

438

450

dbus_interface=client_interface)

439

451

440

452

class DenyCmd(Command):

441

453

def run_on_one_client(self, client, properties):

454

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

455

client.__dbus_object_path__, client_interface)

442

456

client.Approve(dbus.Boolean(False),

443

457

dbus_interface=client_interface)

444

458

536

550

help="Remove client")

537

551

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

538

552

help="Set checker command for client")

539

parser.add_argument("-t", "--timeout",

553

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

540

554

help="Set timeout for client")

541

parser.add_argument("--extended-timeout",

555

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

542

556

help="Set extended timeout for client")

543

parser.add_argument("-i", "--interval",

557

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

544

558

help="Set checker interval for client")

545

559

approve_deny_default = parser.add_mutually_exclusive_group()

546

560

approve_deny_default.add_argument(

551

565

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

552

566

dest="approved_by_default",

553

567

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

554

parser.add_argument("--approval-delay",

568

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

555

569

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

556

parser.add_argument("--approval-duration",

570

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

557

571

help="Set duration of one client approval")

558

572

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

559

573

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

565

579

help="Approve any current client request")

566

580

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

567

581

help="Deny any current client request")

582

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

583

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

568

584

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

569

585

help="Run self-test")

570

586

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

644

660

return commands

645

661

646

662

647

def main():

648

parser = argparse.ArgumentParser()

649

650

add_command_line_options(parser)

651

652

options = parser.parse_args()

663

def check_option_syntax(parser, options):

664

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

665

argparse"""

653

666

654

667

def has_actions(options):

655

668

return any((options.enable,

683

696

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

684

697

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

685

698

699

700

def main():

701

parser = argparse.ArgumentParser()

702

703

add_command_line_options(parser)

704

705

options = parser.parse_args()

706

707

check_option_syntax(parser, options)

708

686

709

clientnames = options.client

687

710

711

if options.debug:

712

log.setLevel(logging.DEBUG)

713

688

714

try:

689

715

bus = dbus.SystemBus()

716

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

717

server_path)

690

718

mandos_dbus_objc = bus.get_object(busname, server_path)

691

719

except dbus.exceptions.DBusException:

692

720

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

705

733

dbus_filter = NullFilter()

706

734

try:

707

735

dbus_logger.addFilter(dbus_filter)

736

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

737

server_path, dbus.OBJECT_MANAGER_IFACE)

708

738

mandos_clients = {path: ifs_and_props[client_interface]

709

739

for path, ifs_and_props in

710

740

mandos_serv_object_manager

1093

1123

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1094

1124

command = SetSecretCmd

1095

1125

property = "Secret"

1096

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

1126

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

1097

1127

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

1098

1128

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

1099

1129

1100

1130

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1101

1131

command = SetTimeoutCmd

1102

1132

property = "Timeout"

1103

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1104

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

1133

values_to_set = [datetime.timedelta(),

1134

datetime.timedelta(minutes=5),

1135

datetime.timedelta(seconds=1),

1136

datetime.timedelta(weeks=1),

1137

datetime.timedelta(weeks=52)]

1138

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

1105

1139

1106

1140

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1107

1141

command = SetExtendedTimeoutCmd

1108

1142

property = "ExtendedTimeout"

1109

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1110

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

1143

values_to_set = [datetime.timedelta(),

1144

datetime.timedelta(minutes=5),

1145

datetime.timedelta(seconds=1),

1146

datetime.timedelta(weeks=1),

1147

datetime.timedelta(weeks=52)]

1148

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

1111

1149

1112

1150

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1113

1151

command = SetIntervalCmd

1114

1152

property = "Interval"

1115

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1116

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

1153

values_to_set = [datetime.timedelta(),

1154

datetime.timedelta(minutes=5),

1155

datetime.timedelta(seconds=1),

1156

datetime.timedelta(weeks=1),

1157

datetime.timedelta(weeks=52)]

1158

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

1117

1159

1118

1160

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1119

1161

command = SetApprovalDelayCmd

1120

1162

property = "ApprovalDelay"

1121

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1122

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

1163

values_to_set = [datetime.timedelta(),

1164

datetime.timedelta(minutes=5),

1165

datetime.timedelta(seconds=1),

1166

datetime.timedelta(weeks=1),

1167

datetime.timedelta(weeks=52)]

1168

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

1123

1169

1124

1170

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1125

1171

command = SetApprovalDurationCmd

1126

1172

property = "ApprovalDuration"

1127

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1128

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

1173

values_to_set = [datetime.timedelta(),

1174

datetime.timedelta(minutes=5),

1175

datetime.timedelta(seconds=1),

1176

datetime.timedelta(weeks=1),

1177

datetime.timedelta(weeks=52)]

1178

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

1129

1179

1130

1180

class Test_command_from_options(unittest.TestCase):

1131

1181

def setUp(self):

1135

1185

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

1136

1186

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

1137

1187

options = self.parser.parse_args(args)

1188

check_option_syntax(self.parser, options)

1138

1189

commands = commands_from_options(options)

1139

1190

self.assertEqual(len(commands), 1)

1140

1191

command = commands[0]

1149

1200

self.assert_command_from_args(["--verbose"], PrintTableCmd,

1150

1201

verbose=True)

1151

1202

1203

def test_print_table_verbose_short(self):

1204

self.assert_command_from_args(["-v"], PrintTableCmd,

1205

verbose=True)

1206

1152

1207

def test_enable(self):

1153

1208

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

1154

1209

1210

def test_enable_short(self):

1211

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

1212

1155

1213

def test_disable(self):

1156

1214

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

1157

1215

DisableCmd)

1158

1216

1217

def test_disable_short(self):

1218

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

1219

1159

1220

def test_bump_timeout(self):

1160

1221

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

1161

1222

BumpTimeoutCmd)

1162

1223

1224

def test_bump_timeout_short(self):

1225

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

1226

1163

1227

def test_start_checker(self):

1164

1228

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

1165

1229

StartCheckerCmd)

1172

1236

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

1173

1237

RemoveCmd)

1174

1238

1239

def test_remove_short(self):

1240

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

1241

1175

1242

def test_checker(self):

1176

1243

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

1177

1244

SetCheckerCmd, value_to_set=":")

1180

1247

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

1181

1248

SetCheckerCmd, value_to_set="")

1182

1249

1250

def test_checker_short(self):

1251

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

1252

SetCheckerCmd, value_to_set=":")

1253

1183

1254

def test_timeout(self):

1184

1255

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

1185

1256

SetTimeoutCmd,

1186

1257

value_to_set=300000)

1187

1258

1259

def test_timeout_short(self):

1260

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

1261

SetTimeoutCmd,

1262

value_to_set=300000)

1263

1188

1264

def test_extended_timeout(self):

1189

1265

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

1190

1266

"foo"],

1196

1272

SetIntervalCmd,

1197

1273

value_to_set=120000)

1198

1274

1275

def test_interval_short(self):

1276

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

1277

SetIntervalCmd,

1278

value_to_set=120000)

1279

1199

1280

def test_approve_by_default(self):

1200

1281

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

1201

1282

ApproveByDefaultCmd)

1219

1300

"foo"], SetHostCmd,

1220

1301

value_to_set="foo.example.org")

1221

1302

1303

def test_host_short(self):

1304

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

1305

"foo"], SetHostCmd,

1306

value_to_set="foo.example.org")

1307

1222

1308

def test_secret_devnull(self):

1223

1309

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

1224

1310

"foo"], SetSecretCmd,

1233

1319

"foo"], SetSecretCmd,

1234

1320

value_to_set=value)

1235

1321

1322

def test_secret_devnull_short(self):

1323

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

1324

SetSecretCmd, value_to_set=b"")

1325

1326

def test_secret_tempfile_short(self):

1327

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1328

value = b"secret\0xyzzy\nbar"

1329

f.write(value)

1330

f.seek(0)

1331

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

1332

SetSecretCmd,

1333

value_to_set=value)

1334

1236

1335

def test_approve(self):

1237

1336

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

1238

1337

ApproveCmd)

1239

1338

1339

def test_approve_short(self):

1340

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

1341

1240

1342

def test_deny(self):

1241

1343

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

1242

1344

1345

def test_deny_short(self):

1346

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

1347

1243

1348

def test_dump_json(self):

1244

1349

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

1245

1350

1247

1352

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

1248

1353

IsEnabledCmd)

1249

1354

1355

def test_is_enabled_short(self):

1356

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

1357

1358

1359

class Test_check_option_syntax(unittest.TestCase):

1360

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

1361

# check_option_syntax()

1362

actions = {

1363

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

1364

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

1365

# never used

1366

"enable": True,

1367

"disable": True,

1368

"bump_timeout": True,

1369

"start_checker": True,

1370

"stop_checker": True,

1371

"is_enabled": True,

1372

"remove": True,

1373

"checker": "x",

1374

"timeout": datetime.timedelta(),

1375

"extended_timeout": datetime.timedelta(),

1376

"interval": datetime.timedelta(),

1377

"approved_by_default": True,

1378

"approval_delay": datetime.timedelta(),

1379

"approval_duration": datetime.timedelta(),

1380

"host": "x",

1381

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

1382

"approve": True,

1383

"deny": True,

1384

}

1385

1386

def setUp(self):

1387

self.parser = argparse.ArgumentParser()

1388

add_command_line_options(self.parser)

1389

1390

@contextlib.contextmanager

1391

def assertParseError(self):

1392

with self.assertRaises(SystemExit) as e:

1393

with self.temporarily_suppress_stderr():

1394

yield

1395

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

1396

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

1397

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

1398

1399

@staticmethod

1400

@contextlib.contextmanager

1401

def temporarily_suppress_stderr():

1402

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

1403

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

1404

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

1405

os.close(null)

1406

try:

1407

yield

1408

finally:

1409

# restore stderr

1410

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

1411

os.close(stderrcopy)

1412

1413

def check_option_syntax(self, options):

1414

check_option_syntax(self.parser, options)

1415

1416

def test_actions_requires_client_or_all(self):

1417

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

1418

options = self.parser.parse_args()

1419

setattr(options, action, value)

1420

with self.assertParseError():

1421

self.check_option_syntax(options)

1422

1423

def test_actions_conflicts_with_verbose(self):

1424

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

1425

options = self.parser.parse_args()

1426

setattr(options, action, value)

1427

options.verbose = True

1428

with self.assertParseError():

1429

self.check_option_syntax(options)

1430

1431

def test_dump_json_conflicts_with_verbose(self):

1432

options = self.parser.parse_args()

1433

options.dump_json = True

1434

options.verbose = True

1435

with self.assertParseError():

1436

self.check_option_syntax(options)

1437

1438

def test_dump_json_conflicts_with_action(self):

1439

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

1440

options = self.parser.parse_args()

1441

setattr(options, action, value)

1442

options.dump_json = True

1443

with self.assertParseError():

1444

self.check_option_syntax(options)

1445

1446

def test_all_can_not_be_alone(self):

1447

options = self.parser.parse_args()

1448

options.all = True

1449

with self.assertParseError():

1450

self.check_option_syntax(options)

1451

1452

def test_all_is_ok_with_any_action(self):

1453

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

1454

options = self.parser.parse_args()

1455

setattr(options, action, value)

1456

options.all = True

1457

self.check_option_syntax(options)

1458

1459

def test_is_enabled_fails_without_client(self):

1460

options = self.parser.parse_args()

1461

options.is_enabled = True

1462

with self.assertParseError():

1463

self.check_option_syntax(options)

1464

1465

def test_is_enabled_works_with_one_client(self):

1466

options = self.parser.parse_args()

1467

options.is_enabled = True

1468

options.client = ["foo"]

1469

self.check_option_syntax(options)

1470

1471

def test_is_enabled_fails_with_two_clients(self):

1472

options = self.parser.parse_args()

1473

options.is_enabled = True

1474

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

1475

with self.assertParseError():

1476

self.check_option_syntax(options)

1477

1250

1478

1251

1479

1252

1480

def should_only_run_tests():