/mandos/trunk : revision 1045

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

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

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

595

611

if options.is_enabled:

596

612

commands.append(IsEnabledCmd())

597

613

598

if options.remove:

599

commands.append(RemoveCmd())

600

601

614

if options.checker is not None:

602

615

commands.append(SetCheckerCmd(options.checker))

603

616

636

649

if options.deny:

637

650

commands.append(DenyCmd())

638

651

652

if options.remove:

653

commands.append(RemoveCmd())

654

639

655

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

640

656

# optionally verbosely

641

657

if not commands:

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,

682

695

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

683

696

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

684

697

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

698

if options.remove:

699

options.remove = False

700

if has_actions(options) and not options.deny:

701

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

702

options.remove = True

703

704

705

def main():

706

parser = argparse.ArgumentParser()

707

708

add_command_line_options(parser)

709

710

options = parser.parse_args()

711

712

check_option_syntax(parser, options)

685

713

686

714

clientnames = options.client

687

715

716

if options.debug:

717

log.setLevel(logging.DEBUG)

718

688

719

try:

689

720

bus = dbus.SystemBus()

721

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

722

server_path)

690

723

mandos_dbus_objc = bus.get_object(busname, server_path)

691

724

except dbus.exceptions.DBusException:

692

725

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

705

738

dbus_filter = NullFilter()

706

739

try:

707

740

dbus_logger.addFilter(dbus_filter)

741

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

742

server_path, dbus.OBJECT_MANAGER_IFACE)

708

743

mandos_clients = {path: ifs_and_props[client_interface]

709

744

for path, ifs_and_props in

710

745

mandos_serv_object_manager

1093

1128

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1094

1129

command = SetSecretCmd

1095

1130

property = "Secret"

1096

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

1131

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

1097

1132

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

1098

1133

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

1099

1134

1155

1190

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

1156

1191

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

1157

1192

options = self.parser.parse_args(args)

1193

check_option_syntax(self.parser, options)

1158

1194

commands = commands_from_options(options)

1159

1195

self.assertEqual(len(commands), 1)

1160

1196

command = commands[0]

1169

1205

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

1170

1206

verbose=True)

1171

1207

1208

def test_print_table_verbose_short(self):

1209

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

1210

verbose=True)

1211

1172

1212

def test_enable(self):

1173

1213

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

1174

1214

1215

def test_enable_short(self):

1216

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

1217

1175

1218

def test_disable(self):

1176

1219

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

1177

1220

DisableCmd)

1178

1221

1222

def test_disable_short(self):

1223

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

1224

1179

1225

def test_bump_timeout(self):

1180

1226

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

1181

1227

BumpTimeoutCmd)

1182

1228

1229

def test_bump_timeout_short(self):

1230

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

1231

1183

1232

def test_start_checker(self):

1184

1233

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

1185

1234

StartCheckerCmd)

1192

1241

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

1193

1242

RemoveCmd)

1194

1243

1244

def test_remove_short(self):

1245

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

1246

1195

1247

def test_checker(self):

1196

1248

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

1197

1249

SetCheckerCmd, value_to_set=":")

1200

1252

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

1201

1253

SetCheckerCmd, value_to_set="")

1202

1254

1255

def test_checker_short(self):

1256

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

1257

SetCheckerCmd, value_to_set=":")

1258

1203

1259

def test_timeout(self):

1204

1260

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

1205

1261

SetTimeoutCmd,

1206

1262

value_to_set=300000)

1207

1263

1264

def test_timeout_short(self):

1265

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

1266

SetTimeoutCmd,

1267

value_to_set=300000)

1268

1208

1269

def test_extended_timeout(self):

1209

1270

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

1210

1271

"foo"],

1216

1277

SetIntervalCmd,

1217

1278

value_to_set=120000)

1218

1279

1280

def test_interval_short(self):

1281

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

1282

SetIntervalCmd,

1283

value_to_set=120000)

1284

1219

1285

def test_approve_by_default(self):

1220

1286

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

1221

1287

ApproveByDefaultCmd)

1239

1305

"foo"], SetHostCmd,

1240

1306

value_to_set="foo.example.org")

1241

1307

1308

def test_host_short(self):

1309

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

1310

"foo"], SetHostCmd,

1311

value_to_set="foo.example.org")

1312

1242

1313

def test_secret_devnull(self):

1243

1314

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

1244

1315

"foo"], SetSecretCmd,

1253

1324

"foo"], SetSecretCmd,

1254

1325

value_to_set=value)

1255

1326

1327

def test_secret_devnull_short(self):

1328

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

1329

SetSecretCmd, value_to_set=b"")

1330

1331

def test_secret_tempfile_short(self):

1332

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

1333

value = b"secret\0xyzzy\nbar"

1334

f.write(value)

1335

f.seek(0)

1336

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

1337

SetSecretCmd,

1338

value_to_set=value)

1339

1256

1340

def test_approve(self):

1257

1341

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

1258

1342

ApproveCmd)

1259

1343

1344

def test_approve_short(self):

1345

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

1346

1260

1347

def test_deny(self):

1261

1348

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

1262

1349

1350

def test_deny_short(self):

1351

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

1352

1263

1353

def test_dump_json(self):

1264

1354

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

1265

1355

1267

1357

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

1268

1358

IsEnabledCmd)

1269

1359

1360

def test_is_enabled_short(self):

1361

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

1362

1363

def test_deny_before_remove(self):

1364

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

1365

check_option_syntax(self.parser, options)

1366

commands = commands_from_options(options)

1367

self.assertEqual(len(commands), 2)

1368

self.assertIsInstance(commands[0], DenyCmd)

1369

self.assertIsInstance(commands[1], RemoveCmd)

1370

1371

def test_deny_before_remove_reversed(self):

1372

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

1373

check_option_syntax(self.parser, options)

1374

commands = commands_from_options(options)

1375

self.assertEqual(len(commands), 2)

1376

self.assertIsInstance(commands[0], DenyCmd)

1377

self.assertIsInstance(commands[1], RemoveCmd)

1378

1379

1380

class Test_check_option_syntax(unittest.TestCase):

1381

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

1382

# check_option_syntax()

1383

actions = {

1384

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

1385

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

1386

# never used

1387

"enable": True,

1388

"disable": True,

1389

"bump_timeout": True,

1390

"start_checker": True,

1391

"stop_checker": True,

1392

"is_enabled": True,

1393

"remove": True,

1394

"checker": "x",

1395

"timeout": datetime.timedelta(),

1396

"extended_timeout": datetime.timedelta(),

1397

"interval": datetime.timedelta(),

1398

"approved_by_default": True,

1399

"approval_delay": datetime.timedelta(),

1400

"approval_duration": datetime.timedelta(),

1401

"host": "x",

1402

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

1403

"approve": True,

1404

"deny": True,

1405

}

1406

1407

def setUp(self):

1408

self.parser = argparse.ArgumentParser()

1409

add_command_line_options(self.parser)

1410

1411

@contextlib.contextmanager

1412

def assertParseError(self):

1413

with self.assertRaises(SystemExit) as e:

1414

with self.temporarily_suppress_stderr():

1415

yield

1416

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

1417

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

1418

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

1419

1420

@staticmethod

1421

@contextlib.contextmanager

1422

def temporarily_suppress_stderr():

1423

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

1424

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

1425

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

1426

os.close(null)

1427

try:

1428

yield

1429

finally:

1430

# restore stderr

1431

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

1432

os.close(stderrcopy)

1433

1434

def check_option_syntax(self, options):

1435

check_option_syntax(self.parser, options)

1436

1437

def test_actions_requires_client_or_all(self):

1438

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

1439

options = self.parser.parse_args()

1440

setattr(options, action, value)

1441

with self.assertParseError():

1442

self.check_option_syntax(options)

1443

1444

def test_actions_conflicts_with_verbose(self):

1445

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

1446

options = self.parser.parse_args()

1447

setattr(options, action, value)

1448

options.verbose = True

1449

with self.assertParseError():

1450

self.check_option_syntax(options)

1451

1452

def test_dump_json_conflicts_with_verbose(self):

1453

options = self.parser.parse_args()

1454

options.dump_json = True

1455

options.verbose = True

1456

with self.assertParseError():

1457

self.check_option_syntax(options)

1458

1459

def test_dump_json_conflicts_with_action(self):

1460

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

1461

options = self.parser.parse_args()

1462

setattr(options, action, value)

1463

options.dump_json = True

1464

with self.assertParseError():

1465

self.check_option_syntax(options)

1466

1467

def test_all_can_not_be_alone(self):

1468

options = self.parser.parse_args()

1469

options.all = True

1470

with self.assertParseError():

1471

self.check_option_syntax(options)

1472

1473

def test_all_is_ok_with_any_action(self):

1474

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

1475

options = self.parser.parse_args()

1476

setattr(options, action, value)

1477

options.all = True

1478

self.check_option_syntax(options)

1479

1480

def test_is_enabled_fails_without_client(self):

1481

options = self.parser.parse_args()

1482

options.is_enabled = True

1483

with self.assertParseError():

1484

self.check_option_syntax(options)

1485

1486

def test_is_enabled_works_with_one_client(self):

1487

options = self.parser.parse_args()

1488

options.is_enabled = True

1489

options.client = ["foo"]

1490

self.check_option_syntax(options)

1491

1492

def test_is_enabled_fails_with_two_clients(self):

1493

options = self.parser.parse_args()

1494

options.is_enabled = True

1495

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

1496

with self.assertParseError():

1497

self.check_option_syntax(options)

1498

1499

def test_remove_can_only_be_combined_with_action_deny(self):

1500

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

1501

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

1502

continue

1503

options = self.parser.parse_args()

1504

setattr(options, action, value)

1505

options.all = True

1506

options.remove = True

1507

with self.assertParseError():

1508

self.check_option_syntax(options)

1509

1270

1510

1271

1511

1272

1512

def should_only_run_tests():