ceph/qa/tasks/cephfs/test_pool_perm.py

   1 from textwrap import dedent
   2 from teuthology.exceptions import CommandFailedError
   3 from tasks.cephfs.cephfs_test_case import CephFSTestCase
   4 import os
   5
   6
   7 class TestPoolPerm(CephFSTestCase):
   8     def test_pool_perm(self):
   9         self.mount_a.run_shell(["touch", "test_file"])
  10
  11         file_path = os.path.join(self.mount_a.mountpoint, "test_file")
  12
  13         remote_script = dedent("""
  14             import os
  15             import errno
  16
  17             fd = os.open("{path}", os.O_RDWR)
  18             try:
  19                 if {check_read}:
  20                     ret = os.read(fd, 1024)
  21                 else:
  22                     os.write(fd, b'content')
  23             except OSError as e:
  24                 if e.errno != errno.EPERM:
  25                     raise
  26             else:
  27                 raise RuntimeError("client does not check permission of data pool")
  28             """)
  29
  30         client_name = "client.{0}".format(self.mount_a.client_id)
  31
  32         # set data pool read only
  33         self.fs.mon_manager.raw_cluster_cmd_result(
  34             'auth', 'caps', client_name, 'mds', 'allow', 'mon', 'allow r', 'osd',
  35             'allow r pool={0}'.format(self.fs.get_data_pool_name()))
  36
  37         self.mount_a.umount_wait()
  38         self.mount_a.mount_wait()
  39
  40         # write should fail
  41         self.mount_a.run_python(remote_script.format(path=file_path, check_read=str(False)))
  42
  43         # set data pool write only
  44         self.fs.mon_manager.raw_cluster_cmd_result(
  45             'auth', 'caps', client_name, 'mds', 'allow', 'mon', 'allow r', 'osd',
  46             'allow w pool={0}'.format(self.fs.get_data_pool_name()))
  47
  48         self.mount_a.umount_wait()
  49         self.mount_a.mount_wait()
  50
  51         # read should fail
  52         self.mount_a.run_python(remote_script.format(path=file_path, check_read=str(True)))
  53
  54     def test_forbidden_modification(self):
  55         """
  56         That a client who does not have the capability for setting
  57         layout pools is prevented from doing so.
  58         """
  59
  60         # Set up
  61         client_name = "client.{0}".format(self.mount_a.client_id)
  62         new_pool_name = "data_new"
  63         self.fs.add_data_pool(new_pool_name)
  64
  65         self.mount_a.run_shell(["touch", "layoutfile"])
  66         self.mount_a.run_shell(["mkdir", "layoutdir"])
  67
  68         # Set MDS 'rw' perms: missing 'p' means no setting pool layouts
  69         self.fs.mon_manager.raw_cluster_cmd_result(
  70             'auth', 'caps', client_name, 'mds', 'allow rw', 'mon', 'allow r',
  71             'osd',
  72             'allow rw pool={0},allow rw pool={1}'.format(
  73                 self.fs.get_data_pool_names()[0],
  74                 self.fs.get_data_pool_names()[1],
  75             ))
  76
  77         self.mount_a.umount_wait()
  78         self.mount_a.mount_wait()
  79
  80         with self.assertRaises(CommandFailedError):
  81             self.mount_a.setfattr("layoutfile", "ceph.file.layout.pool",
  82                                   new_pool_name)
  83         with self.assertRaises(CommandFailedError):
  84             self.mount_a.setfattr("layoutdir", "ceph.dir.layout.pool",
  85                                   new_pool_name)
  86         self.mount_a.umount_wait()
  87
  88         # Set MDS 'rwp' perms: should now be able to set layouts
  89         self.fs.mon_manager.raw_cluster_cmd_result(
  90             'auth', 'caps', client_name, 'mds', 'allow rwp', 'mon', 'allow r',
  91             'osd',
  92             'allow rw pool={0},allow rw pool={1}'.format(
  93                 self.fs.get_data_pool_names()[0],
  94                 self.fs.get_data_pool_names()[1],
  95             ))
  96         self.mount_a.mount_wait()
  97         self.mount_a.setfattr("layoutfile", "ceph.file.layout.pool",
  98                               new_pool_name)
  99         self.mount_a.setfattr("layoutdir", "ceph.dir.layout.pool",
 100                               new_pool_name)
 101         self.mount_a.umount_wait()
 102
 103     def tearDown(self):
 104         self.fs.mon_manager.raw_cluster_cmd_result(
 105             'auth', 'caps', "client.{0}".format(self.mount_a.client_id),
 106             'mds', 'allow', 'mon', 'allow r', 'osd',
 107             'allow rw pool={0}'.format(self.fs.get_data_pool_names()[0]))
 108         super(TestPoolPerm, self).tearDown()
 109