CA-387770: check for read-only shared fs at create

Signed-off-by: Mark Syms <[email protected]>

drivers/NFSSR.py

@@ -225,9 +225,14 @@ def create(self, sr_uuid, size):
                 except util.CommandException as inst:
                     if inst.code != errno.EEXIST:
                         self.detach(sr_uuid)
-                        raise xs_errors.XenError('NFSCreate',
-                            opterr='remote directory creation error is %d'
-                            % inst.code)
+                        if inst.code == errno.EROFS:
+                            raise xs_errors.XenError('SharedFileSystemNoWrite',
+                                opterr='remote filesystem is read-only error is %d'
+                                % inst.code)
+                        else:
+                            raise xs_errors.XenError('NFSCreate',
+                                opterr='remote directory creation error is %d'
+                                % inst.code)
         self.detach(sr_uuid)
 
     def delete(self, sr_uuid):

drivers/SMBSR.py

@@ -272,11 +272,17 @@ def create(self, sr_uuid, size):
                         self.unmount(self.mountpoint, True)
                     except SMBException:
                         util.logException('SMBSR.unmount()')
-                    raise xs_errors.XenError(
+
+                    if inst.code in [errno.EROFS, errno.EPERM, errno.EACCES]:
+                        raise xs_errors.XenError(
+                            'SharedFileSystemNoWrite',
+                            opterr='remote filesystem is read-only error is %d'
+                                   % inst.code) from inst
+                    else:
+                        raise xs_errors.XenError(
                             'SMBCreate',
                             opterr="remote directory creation error: {}"
-                                    .format(os.strerror(inst.code))
-                    )
+                            .format(os.strerror(inst.code))) from inst
         self.detach(sr_uuid)
 
     def delete(self, sr_uuid):

tests/test_NFSSR.py

@@ -1,10 +1,13 @@
+import errno
 import unittest.mock as mock
 import nfs
 import NFSSR
 import SR
 import unittest
 from uuid import uuid4
 
+import util
+
 
 class FakeNFSSR(NFSSR.NFSSR):
     uuid = None
@@ -82,6 +85,62 @@ def test_sr_create(self, validate_nfsversion, check_server_tcp, _testhost,
         size = 100
         nfssr.create(sr_uuid, size)
 
+    @mock.patch('util.makedirs')
+    @mock.patch('NFSSR.Lock', autospec=True)
+    @mock.patch('nfs.soft_mount')
+    @mock.patch('util._testHost')
+    @mock.patch('nfs.check_server_tcp')
+    @mock.patch('nfs.validate_nfsversion')
+    @mock.patch('SR.xs_errors.XML_DEFS', "drivers/XE_SR_ERRORCODES.xml")
+    def test_sr_create_readonly(self, validate_nfsversion, check_server_tcp, _testhost,
+                       soft_mount, Lock, makedirs):
+        # Arrange
+        nfssr = self.create_nfssr(server='aServer', serverpath='/aServerpath',
+                                  sr_uuid='UUID', useroptions='options')
+
+        sr_uuid = str(uuid4())
+        size = 100
+
+        def mock_makedirs(path):
+            raise util.CommandException(errno.EROFS)
+
+        makedirs.side_effect = mock_makedirs
+
+        # Act
+        with self.assertRaises(SR.SROSError) as srose:
+            nfssr.create(sr_uuid, size)
+
+        self.assertEqual(srose.exception.errno, 461)
+
+    @mock.patch('util.makedirs')
+    @mock.patch('NFSSR.Lock', autospec=True)
+    @mock.patch('nfs.soft_mount')
+    @mock.patch('util._testHost')
+    @mock.patch('nfs.check_server_tcp')
+    @mock.patch('nfs.validate_nfsversion')
+    @mock.patch('SR.xs_errors.XML_DEFS', "drivers/XE_SR_ERRORCODES.xml")
+    def test_sr_create_noperm(self, validate_nfsversion, check_server_tcp, _testhost,
+                       soft_mount, Lock, makedirs):
+        # Arrange
+        nfssr = self.create_nfssr(server='aServer', serverpath='/aServerpath',
+                                  sr_uuid='UUID', useroptions='options')
+
+        sr_uuid = str(uuid4())
+        size = 100
+
+        def mock_makedirs(path):
+            raise util.CommandException(errno.EPERM)
+
+
+        makedirs.side_effect = mock_makedirs
+
+        # Act
+        with self.assertRaises(SR.SROSError) as srose:
+            nfssr.create(sr_uuid, size)
+
+        self.assertEqual(srose.exception.errno, 88)
+
+
     @mock.patch('NFSSR.os.rmdir')
     @mock.patch('NFSSR.Lock', autospec=True)
     @mock.patch('nfs.soft_mount')

tests/test_SMBSR.py

@@ -1,10 +1,13 @@
 import unittest
 import unittest.mock as mock
+import uuid
+
 import SR
 import SMBSR
 import testlib
 import util
 import errno
+import XenAPI
 
 
 class FakeSMBSR(SMBSR.SMBSR):
@@ -29,6 +32,26 @@ def __init__(self, srcmd, none):
 
 class Test_SMBSR(unittest.TestCase):
 
+    def setUp(self):
+        self.addCleanup(mock.patch.stopall)
+
+        pread_patcher = mock.patch('SMBSR.util.pread', autospec=True)
+        self.mock_pread = pread_patcher.start()
+        self.mock_pread.side_effect = self.pread
+        self.pread_results = {}
+
+        listdir_patcher = mock.patch('SMBSR.util.listdir', autospec=True)
+        self.mock_list_dir = listdir_patcher.start()
+
+        rmdir_patcher = mock.patch('SMBSR.os.rmdir', autospec=True)
+        self.mock_rmdir = rmdir_patcher.start()
+
+    def arg_key(self, args):
+        return ':'.join(args)
+
+    def pread(self, args, new_env=None, quiet=False, text=False):
+        return self.pread_results.get(self.arg_key(args), None)
+
     def create_smbsr(self, sr_uuid='asr_uuid', server='\\aServer', serverpath='/aServerpath', username='aUsername', password='aPassword', dconf_update={}):
         srcmd = mock.Mock()
         srcmd.dconf = {
@@ -75,56 +98,51 @@ def test_attach_if_mounted_then_attached(self, mock_lock, mock_checkmount):
     @mock.patch('SMBSR.SMBSR.checkmount', autospec=True)
     @mock.patch('SMBSR.SMBSR.makeMountPoint', autospec=True)
     @mock.patch('SMBSR.Lock', autospec=True)
-    @mock.patch('util.pread', autospec=True)
     @mock.patch('os.symlink', autospec=True)
-    @mock.patch('util.listdir', autospec=True)
-    def test_attach_vanilla(self, listdir, symlink, pread, mock_lock,
+    def test_attach_vanilla(self, symlink, mock_lock,
                             makeMountPoint, mock_checkmount, mock_checklinks,
                             mock_checkwritable):
         mock_checkmount.return_value = False
         smbsr = self.create_smbsr()
         makeMountPoint.return_value = "/var/mount"
         smbsr.attach('asr_uuid')
         self.assertTrue(smbsr.attached)
-        pread.assert_called_with(['mount.cifs', '\\aServer', "/var/mount", '-o', 'cache=loose,vers=3.0,actimeo=0'],
-                                 new_env={'PASSWD': 'aPassword', 'USER': 'aUsername'})
+        self.mock_pread.assert_called_with(
+            ['mount.cifs', '\\aServer', "/var/mount", '-o', 'cache=loose,vers=3.0,actimeo=0'],
+            new_env={'PASSWD': 'aPassword', 'USER': 'aUsername'})
 
     @mock.patch('FileSR.SharedFileSR._check_writable', autospec=True)
     @mock.patch('FileSR.SharedFileSR._check_hardlinks', autospec=True)
     @mock.patch('SMBSR.SMBSR.checkmount', autospec=True)
     @mock.patch('SMBSR.SMBSR.makeMountPoint', autospec=True)
     @mock.patch('SMBSR.Lock', autospecd=True)
-    @mock.patch('util.pread', autospec=True)
     @mock.patch('os.symlink', autospec=True)
-    @mock.patch('util.listdir', autospec=True)
     def test_attach_with_cifs_password(
-            self, listdir, symlink, pread, mock_lock, makeMountPoint,
+            self, symlink, mock_lock, makeMountPoint,
             mock_checkmount, mock_checklinks, mock_checkwritable):
         smbsr = self.create_smbsr(dconf_update={"password": "winter2019"})
         mock_checkmount.return_value = False
         makeMountPoint.return_value = "/var/mount"
         smbsr.attach('asr_uuid')
         self.assertTrue(smbsr.attached)
-        pread.assert_called_with(['mount.cifs', '\\aServer', "/var/mount", '-o', 'cache=loose,vers=3.0,actimeo=0'], new_env={'PASSWD': 'winter2019', 'USER': 'aUsername'})
+        self.mock_pread.assert_called_with(['mount.cifs', '\\aServer', "/var/mount", '-o', 'cache=loose,vers=3.0,actimeo=0'], new_env={'PASSWD': 'winter2019', 'USER': 'aUsername'})
 
     @mock.patch('FileSR.SharedFileSR._check_writable', autospec=True)
     @mock.patch('FileSR.SharedFileSR._check_hardlinks', autospec=True)
     @mock.patch('SMBSR.SMBSR.checkmount', autospec=True)
     @mock.patch('SMBSR.SMBSR.makeMountPoint', autospec=True)
     @mock.patch('SMBSR.Lock', autospecd=True)
-    @mock.patch('util.pread', autospec=True)
     @mock.patch('os.symlink', autospec=True)
-    @mock.patch('util.listdir', autospec=True)
     def test_attach_with_cifs_password_and_domain(
-            self, listdir, symlink, pread, mock_lock, makeMountPoint,
+            self, symlink, mock_lock, makeMountPoint,
             mock_checkmount, mock_checklinks, mock_checkwritable):
         smbsr = self.create_smbsr(username="citrix\jsmith", dconf_update={"password": "winter2019"})
         mock_checkmount.return_value = False
         makeMountPoint.return_value = "/var/mount"
         smbsr.attach('asr_uuid')
         self.assertTrue(smbsr.attached)
         # We mocked listdir as this calls pread and assert_called_with only records the last call.
-        pread.assert_called_with(['mount.cifs', '\\aServer', "/var/mount", '-o', 'cache=loose,vers=3.0,actimeo=0,domain=citrix'], new_env={'PASSWD': 'winter2019', 'USER': 'jsmith'})
+        self.mock_pread.assert_called_with(['mount.cifs', '\\aServer', "/var/mount", '-o', 'cache=loose,vers=3.0,actimeo=0,domain=citrix'], new_env={'PASSWD': 'winter2019', 'USER': 'jsmith'})
 
     @mock.patch('FileSR.SharedFileSR._check_writable', autospec=True)
     @mock.patch('FileSR.SharedFileSR._check_hardlinks', autospec=True)
@@ -241,3 +259,83 @@ def test_mount_mountpoint_isdir(self, mock_time, mock_lock, mock_isdir):
     def test_mount_mountpoint_empty_string(self, mock_lock):
         smbsr = self.create_smbsr()
         self.assertRaises(SMBSR.SMBException, smbsr.mount, "")
+
+    @mock.patch('util.makedirs', autospec=True)
+    @mock.patch('util.get_pool_restrictions', autospec=True)
+    @mock.patch('SMBSR.Lock', autospec=True)
+    @mock.patch('SMBSR.os.symlink', autospec=True)
+    def test_create_success(self, symlink, lock, restrict, makedirs):
+        # Arrange
+        smbsr = self.create_smbsr()
+        smbsr.session = mock.MagicMock(spec=XenAPI)
+        restrict.return_value = []
+        sr_uuid = str(uuid.uuid4())
+        self.mock_list_dir.return_value = []
+
+        # Act
+        smbsr.create(sr_uuid, 10 * 1024 * 1024 * 1024)
+
+        # Assert
+        self.mock_pread.assert_called_with(
+            ['mount.cifs', '\\aServer', "/var/run/sr-mount/SMB/Server/asr_uuid",
+             '-o', 'cache=loose,vers=3.0,actimeo=0'],
+            new_env={'USER': 'aUsername', 'PASSWD': 'aPassword'})
+
+    @mock.patch('util.makedirs', autospec=True)
+    @mock.patch('util.get_pool_restrictions', autospec=True)
+    @mock.patch('SMBSR.Lock', autospec=True)
+    @mock.patch('SMBSR.os.symlink', autospec=True)
+    @mock.patch('SR.xs_errors.XML_DEFS', "drivers/XE_SR_ERRORCODES.xml")
+    def test_create_read_only(self, symlink, lock, restrict, makedirs):
+        # Arrange
+        smbsr = self.create_smbsr()
+        smbsr.session = mock.MagicMock(spec=XenAPI)
+        restrict.return_value = []
+        sr_uuid = str(uuid.uuid4())
+        self.mock_list_dir.return_value = []
+
+        def mock_makedirs(path):
+            if path == '/var/run/sr-mount/SMB/Server/asr_uuid':
+                return
+
+            raise util.CommandException(errno.EACCES)
+
+        makedirs.side_effect = mock_makedirs
+
+        # Act
+        with self.assertRaises(SR.SROSError) as srose:
+            smbsr.create(sr_uuid, 10 * 1024 * 1024 * 1024)
+
+        # Assert
+        self.assertEqual(srose.exception.errno, 461)
+        symlink.assert_not_called()
+
+
+    @mock.patch('util.makedirs', autospec=True)
+    @mock.patch('util.get_pool_restrictions', autospec=True)
+    @mock.patch('SMBSR.Lock', autospec=True)
+    @mock.patch('SMBSR.os.symlink', autospec=True)
+    @mock.patch('SR.xs_errors.XML_DEFS', "drivers/XE_SR_ERRORCODES.xml")
+    def test_create_nospace(self, symlink, lock, restrict, makedirs):
+        # Arrange
+        smbsr = self.create_smbsr()
+        smbsr.session = mock.MagicMock(spec=XenAPI)
+        restrict.return_value = []
+        sr_uuid = str(uuid.uuid4())
+        self.mock_list_dir.return_value = []
+
+        def mock_makedirs(path):
+            if path == '/var/run/sr-mount/SMB/Server/asr_uuid':
+                return
+
+            raise util.CommandException(errno.ENOSPC)
+
+        makedirs.side_effect = mock_makedirs
+
+        # Act
+        with self.assertRaises(SR.SROSError) as srose:
+            smbsr.create(sr_uuid, 10 * 1024 * 1024 * 1024)
+
+        # Assert
+        self.assertEqual(srose.exception.errno, 116)
+        symlink.assert_not_called()