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