/mandos/trunk

To get this branch, use:
bzr branch http://bzr.recompile.se/loggerhead/mandos/trunk

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-09 00:35:00 UTC
  • Revision ID: teddy@recompile.se-20190309003500-t0pdqxwy5g8jrfax
mandos-ctl: Make --deny always apply before --remove

* mandos-ctl (commands_from_options): Move options.remove clause to
                                      after options.deny.
  (Test_command_from_options.test_deny_before_remove): New.
  (Test_command_from_options.test_deny_before_remove_reversed): - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
44
44
import logging
45
45
import io
46
46
import tempfile
 
47
import contextlib
47
48
 
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:
645
661
 
646
662
 
647
663
def check_option_syntax(parser, options):
 
664
    """Apply additional restrictions on options, not expressible in
 
665
argparse"""
648
666
 
649
667
    def has_actions(options):
650
668
        return any((options.enable,
690
708
 
691
709
    clientnames = options.client
692
710
 
 
711
    if options.debug:
 
712
        log.setLevel(logging.DEBUG)
 
713
 
693
714
    try:
694
715
        bus = dbus.SystemBus()
 
716
        log.debug("D-Bus: Connect to: (name=%r, path=%r)", busname,
 
717
                  server_path)
695
718
        mandos_dbus_objc = bus.get_object(busname, server_path)
696
719
    except dbus.exceptions.DBusException:
697
720
        log.critical("Could not connect to Mandos server")
710
733
    dbus_filter = NullFilter()
711
734
    try:
712
735
        dbus_logger.addFilter(dbus_filter)
 
736
        log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,
 
737
                  server_path, dbus.OBJECT_MANAGER_IFACE)
713
738
        mandos_clients = {path: ifs_and_props[client_interface]
714
739
                          for path, ifs_and_props in
715
740
                          mandos_serv_object_manager
1098
1123
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1099
1124
    command = SetSecretCmd
1100
1125
    property = "Secret"
1101
 
    values_to_set = [open("/dev/null", "rb"),
 
1126
    values_to_set = [io.BytesIO(b""),
1102
1127
                     io.BytesIO(b"secret\0xyzzy\nbar")]
1103
1128
    values_to_get = [b"", b"secret\0xyzzy\nbar"]
1104
1129
 
1330
1355
    def test_is_enabled_short(self):
1331
1356
        self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1332
1357
 
 
1358
    def test_deny_before_remove(self):
 
1359
        options = self.parser.parse_args(["--deny", "--remove", "foo"])
 
1360
        check_option_syntax(self.parser, options)
 
1361
        commands = commands_from_options(options)
 
1362
        self.assertEqual(len(commands), 2)
 
1363
        self.assertIsInstance(commands[0], DenyCmd)
 
1364
        self.assertIsInstance(commands[1], RemoveCmd)
 
1365
 
 
1366
    def test_deny_before_remove_reversed(self):
 
1367
        options = self.parser.parse_args(["--remove", "--deny", "--all"])
 
1368
        check_option_syntax(self.parser, options)
 
1369
        commands = commands_from_options(options)
 
1370
        self.assertEqual(len(commands), 2)
 
1371
        self.assertIsInstance(commands[0], DenyCmd)
 
1372
        self.assertIsInstance(commands[1], RemoveCmd)
 
1373
 
 
1374
 
 
1375
class Test_check_option_syntax(unittest.TestCase):
 
1376
    # This mostly corresponds to the definition from has_actions() in
 
1377
    # check_option_syntax()
 
1378
    actions = {
 
1379
        # The actual values set here are not that important, but we do
 
1380
        # at least stick to the correct types, even though they are
 
1381
        # never used
 
1382
        "enable": True,
 
1383
        "disable": True,
 
1384
        "bump_timeout": True,
 
1385
        "start_checker": True,
 
1386
        "stop_checker": True,
 
1387
        "is_enabled": True,
 
1388
        "remove": True,
 
1389
        "checker": "x",
 
1390
        "timeout": datetime.timedelta(),
 
1391
        "extended_timeout": datetime.timedelta(),
 
1392
        "interval": datetime.timedelta(),
 
1393
        "approved_by_default": True,
 
1394
        "approval_delay": datetime.timedelta(),
 
1395
        "approval_duration": datetime.timedelta(),
 
1396
        "host": "x",
 
1397
        "secret": io.BytesIO(b"x"),
 
1398
        "approve": True,
 
1399
        "deny": True,
 
1400
    }
 
1401
 
 
1402
    def setUp(self):
 
1403
        self.parser = argparse.ArgumentParser()
 
1404
        add_command_line_options(self.parser)
 
1405
 
 
1406
    @contextlib.contextmanager
 
1407
    def assertParseError(self):
 
1408
        with self.assertRaises(SystemExit) as e:
 
1409
            with self.temporarily_suppress_stderr():
 
1410
                yield
 
1411
        # Exit code from argparse is guaranteed to be "2".  Reference:
 
1412
        # https://docs.python.org/3/library/argparse.html#exiting-methods
 
1413
        self.assertEqual(e.exception.code, 2)
 
1414
 
 
1415
    @staticmethod
 
1416
    @contextlib.contextmanager
 
1417
    def temporarily_suppress_stderr():
 
1418
        null = os.open(os.path.devnull, os.O_RDWR)
 
1419
        stderrcopy = os.dup(sys.stderr.fileno())
 
1420
        os.dup2(null, sys.stderr.fileno())
 
1421
        os.close(null)
 
1422
        try:
 
1423
            yield
 
1424
        finally:
 
1425
            # restore stderr
 
1426
            os.dup2(stderrcopy, sys.stderr.fileno())
 
1427
            os.close(stderrcopy)
 
1428
 
 
1429
    def check_option_syntax(self, options):
 
1430
        check_option_syntax(self.parser, options)
 
1431
 
 
1432
    def test_actions_requires_client_or_all(self):
 
1433
        for action, value in self.actions.items():
 
1434
            options = self.parser.parse_args()
 
1435
            setattr(options, action, value)
 
1436
            with self.assertParseError():
 
1437
                self.check_option_syntax(options)
 
1438
 
 
1439
    def test_actions_conflicts_with_verbose(self):
 
1440
        for action, value in self.actions.items():
 
1441
            options = self.parser.parse_args()
 
1442
            setattr(options, action, value)
 
1443
            options.verbose = True
 
1444
            with self.assertParseError():
 
1445
                self.check_option_syntax(options)
 
1446
 
 
1447
    def test_dump_json_conflicts_with_verbose(self):
 
1448
        options = self.parser.parse_args()
 
1449
        options.dump_json = True
 
1450
        options.verbose = True
 
1451
        with self.assertParseError():
 
1452
            self.check_option_syntax(options)
 
1453
 
 
1454
    def test_dump_json_conflicts_with_action(self):
 
1455
        for action, value in self.actions.items():
 
1456
            options = self.parser.parse_args()
 
1457
            setattr(options, action, value)
 
1458
            options.dump_json = True
 
1459
            with self.assertParseError():
 
1460
                self.check_option_syntax(options)
 
1461
 
 
1462
    def test_all_can_not_be_alone(self):
 
1463
        options = self.parser.parse_args()
 
1464
        options.all = True
 
1465
        with self.assertParseError():
 
1466
            self.check_option_syntax(options)
 
1467
 
 
1468
    def test_all_is_ok_with_any_action(self):
 
1469
        for action, value in self.actions.items():
 
1470
            options = self.parser.parse_args()
 
1471
            setattr(options, action, value)
 
1472
            options.all = True
 
1473
            self.check_option_syntax(options)
 
1474
 
 
1475
    def test_is_enabled_fails_without_client(self):
 
1476
        options = self.parser.parse_args()
 
1477
        options.is_enabled = True
 
1478
        with self.assertParseError():
 
1479
            self.check_option_syntax(options)
 
1480
 
 
1481
    def test_is_enabled_works_with_one_client(self):
 
1482
        options = self.parser.parse_args()
 
1483
        options.is_enabled = True
 
1484
        options.client = ["foo"]
 
1485
        self.check_option_syntax(options)
 
1486
 
 
1487
    def test_is_enabled_fails_with_two_clients(self):
 
1488
        options = self.parser.parse_args()
 
1489
        options.is_enabled = True
 
1490
        options.client = ["foo", "barbar"]
 
1491
        with self.assertParseError():
 
1492
            self.check_option_syntax(options)
 
1493
 
1333
1494
 
1334
1495
 
1335
1496
def should_only_run_tests():