NFS Security Enhancements: allows secure NFS environment setup

This patch allows an OpenStack environment to run as a secure NAS
environment from the client and server perspective, including having
root squash enabled and not running file operations as the 'root'
user. This also sets Cinder file permissions as 660: removing
other/world file access.

The "nas_secure_file_permissions" option controls the setting of file
permissions when Cinder volumes are created. The option defaults to
"auto" to gracefully handle upgrade scenarios. When set to "auto",
a check is done during Cinder startup to determine if there are
existing Cinder volumes: no volumes will set the option to 'true',
and use secure file permissions. The detection of existing volumes will
set the option to 'false', and use the current insecure method of
handling file permissions.

The "nas_secure_file_operations" option controls whether file
operations are run as the 'root' user or the current OpenStack
'process' user. The option defaults to "auto" to gracefully handle
upgrade scenarios. When set to "auto", a check is done during Cinder
startup to determine if there are existing Cinder volumes: no volumes
will set the option to 'true', be secure and do NOT run as the 'root'
user. The detection of existing volumes will set the option to 'false',
and use the current method of running operations as the 'root' user.
For new installations, a 'marker file' is written so that subsequent
restarts of Cinder will know what the original determination had been.

This patch enables this functionality only for the NFS driver.
Other similar drivers can use this code to enable the same
functionality with the same config options.

DocImpact
Change-Id: I3d25f593beab7f5462576b14ab62d13d8c53e7c6
Implements: blueprint secure-nfs
Partial-Bug: 1260679

cinder/brick/initiator/connector.py

@@ -129,12 +129,12 @@ class InitiatorConnector(executor.Executor):
                    dict(protocol=protocol))
             raise ValueError(msg)
 
-    def check_valid_device(self, path):
+    def check_valid_device(self, path, run_as_root=True):
         cmd = ('dd', 'if=%(path)s' % {"path": path},
                'of=/dev/null', 'count=1')
         out, info = None, None
         try:
-            out, info = self._execute(*cmd, run_as_root=True,
+            out, info = self._execute(*cmd, run_as_root=run_as_root,
                                       root_helper=self._root_helper)
         except putils.ProcessExecutionError as e:
             LOG.error(_("Failed to access the device on the path "

cinder/image/image_utils.py

@@ -52,16 +52,16 @@ CONF = cfg.CONF
 CONF.register_opts(image_helper_opt)
 
 
-def qemu_img_info(path):
+def qemu_img_info(path, run_as_root=True):
-    """Return an object containing the parsed output from qemu-img info."""
+    """Return a object containing the parsed output from qemu-img info."""
     cmd = ('env', 'LC_ALL=C', 'qemu-img', 'info', path)
     if os.name == 'nt':
         cmd = cmd[2:]
-    out, err = utils.execute(*cmd, run_as_root=True)
+    out, err = utils.execute(*cmd, run_as_root=run_as_root)
     return imageutils.QemuImgInfo(out)
 
 
-def convert_image(source, dest, out_format, bps_limit=None):
+def convert_image(source, dest, out_format, bps_limit=None, run_as_root=True):
     """Convert image to other format."""
 
     cmd = ('qemu-img', 'convert',
@@ -88,7 +88,7 @@ def convert_image(source, dest, out_format, bps_limit=None):
     cgcmd = volume_utils.setup_blkio_cgroup(source, dest, bps_limit)
     if cgcmd:
         cmd = tuple(cgcmd) + cmd
-    utils.execute(*cmd, run_as_root=True)
+    utils.execute(*cmd, run_as_root=run_as_root)
 
     duration = timeutils.delta_seconds(start_time, timeutils.utcnow())
 
@@ -142,12 +142,13 @@ def fetch(context, image_service, image_id, path, _user_id, _project_id):
 
 
 def fetch_verify_image(context, image_service, image_id, dest,
-                       user_id=None, project_id=None, size=None):
+                       user_id=None, project_id=None, size=None,
+                       run_as_root=True):
     fetch(context, image_service, image_id, dest,
           None, None)
 
     with fileutils.remove_path_on_error(dest):
-        data = qemu_img_info(dest)
+        data = qemu_img_info(dest, run_as_root=run_as_root)
         fmt = data.file_format
         if fmt is None:
             raise exception.ImageUnacceptable(
@@ -173,21 +174,24 @@ def fetch_verify_image(context, image_service, image_id, dest,
 
 def fetch_to_vhd(context, image_service,
                  image_id, dest, blocksize,
-                 user_id=None, project_id=None):
+                 user_id=None, project_id=None, run_as_root=True):
     fetch_to_volume_format(context, image_service, image_id, dest, 'vpc',
-                           blocksize, user_id, project_id)
+                           blocksize, user_id, project_id,
+                           run_as_root=run_as_root)
 
 
 def fetch_to_raw(context, image_service,
                  image_id, dest, blocksize,
-                 user_id=None, project_id=None, size=None):
+                 user_id=None, project_id=None, size=None, run_as_root=True):
     fetch_to_volume_format(context, image_service, image_id, dest, 'raw',
-                           blocksize, user_id, project_id, size)
+                           blocksize, user_id, project_id, size,
+                           run_as_root=run_as_root)
 
 
 def fetch_to_volume_format(context, image_service,
                            image_id, dest, volume_format, blocksize,
-                           user_id=None, project_id=None, size=None):
+                           user_id=None, project_id=None, size=None,
+                           run_as_root=True):
     if (CONF.image_conversion_dir and not
             os.path.exists(CONF.image_conversion_dir)):
         os.makedirs(CONF.image_conversion_dir)
@@ -208,7 +212,7 @@ def fetch_to_volume_format(context, image_service,
         # whole function.
         try:
             # Use the empty tmp file to make sure qemu_img_info works.
-            qemu_img_info(tmp)
+            qemu_img_info(tmp, run_as_root=run_as_root)
         except processutils.ProcessExecutionError:
             qemu_img = False
             if image_meta:
@@ -241,7 +245,7 @@ def fetch_to_volume_format(context, image_service,
             volume_utils.copy_volume(tmp, dest, image_meta['size'], blocksize)
             return
 
-        data = qemu_img_info(tmp)
+        data = qemu_img_info(tmp, run_as_root=run_as_root)
         virt_size = data.virtual_size / units.Gi
 
         # NOTE(xqueralt): If the image virtual size doesn't fit in the
@@ -276,9 +280,10 @@ def fetch_to_volume_format(context, image_service,
         LOG.debug("%s was %s, converting to %s " % (image_id, fmt,
                                                     volume_format))
         convert_image(tmp, dest, volume_format,
-                      bps_limit=CONF.volume_copy_bps_limit)
+                      bps_limit=CONF.volume_copy_bps_limit,
+                      run_as_root=run_as_root)
 
-        data = qemu_img_info(dest)
+        data = qemu_img_info(dest, run_as_root=run_as_root)
         if data.file_format != volume_format:
             raise exception.ImageUnacceptable(
                 image_id=image_id,
@@ -289,7 +294,7 @@ def fetch_to_volume_format(context, image_service,
 
 
 def upload_volume(context, image_service, image_meta, volume_path,
-                  volume_format='raw'):
+                  volume_format='raw', run_as_root=True):
     image_id = image_meta['id']
     if (image_meta['disk_format'] == volume_format):
         LOG.debug("%s was %s, no need to convert to %s" %
@@ -313,9 +318,10 @@ def upload_volume(context, image_service, image_meta, volume_path,
         LOG.debug("%s was %s, converting to %s" %
                   (image_id, volume_format, image_meta['disk_format']))
         convert_image(volume_path, tmp, image_meta['disk_format'],
-                      bps_limit=CONF.volume_copy_bps_limit)
+                      bps_limit=CONF.volume_copy_bps_limit,
+                      run_as_root=run_as_root)
 
-        data = qemu_img_info(tmp)
+        data = qemu_img_info(tmp, run_as_root=run_as_root)
         if data.file_format != image_meta['disk_format']:
             raise exception.ImageUnacceptable(
                 image_id=image_id,

cinder/tests/test_coraid.py

@@ -832,7 +832,7 @@ class CoraidDriverImageTestCases(CoraidDriverTestCase):
             .connect_volume(self.fake_connection['data'])\
             .AndReturn({'path': self.fake_dev_path})
 
-        aoe_initiator.check_valid_device(self.fake_dev_path)\
+        aoe_initiator.check_valid_device(self.fake_dev_path, mox.IgnoreArg())\
             .AndReturn(True)
 
         aoe_initiator.disconnect_volume(

cinder/tests/test_glusterfs.py

@@ -98,6 +98,8 @@ class GlusterFsDriverTestCase(test.TestCase):
             self.TEST_MNT_POINT_BASE
         self._configuration.glusterfs_sparsed_volumes = True
         self._configuration.glusterfs_qcow2_volumes = False
+        self._configuration.nas_secure_file_permissions = 'false'
+        self._configuration.nas_secure_file_operations = 'false'
 
         self.stubs = stubout.StubOutForTesting()
         self._driver =\

cinder/tests/test_image_utils.py

@@ -71,11 +71,12 @@ class TestUtils(test.TestCase):
         TEST_IMG_SIZE_IN_GB = 1
 
         utils.execute('qemu-img', 'resize', TEST_IMG_SOURCE,
-                      '%sG' % TEST_IMG_SIZE_IN_GB, run_as_root=False)
+                      '%sG' % TEST_IMG_SIZE_IN_GB, run_as_root=True)
 
         mox.ReplayAll()
 
-        image_utils.resize_image(TEST_IMG_SOURCE, TEST_IMG_SIZE_IN_GB)
+        image_utils.resize_image(TEST_IMG_SOURCE, TEST_IMG_SIZE_IN_GB,
+                                 run_as_root=True)
 
         mox.VerifyAll()
 
@@ -100,7 +101,8 @@ class TestUtils(test.TestCase):
 
         mox.ReplayAll()
 
-        image_utils.convert_image(TEST_SOURCE, TEST_DEST, TEST_OUT_FORMAT)
+        image_utils.convert_image(TEST_SOURCE, TEST_DEST, TEST_OUT_FORMAT,
+                                  run_as_root=True)
 
         mox.VerifyAll()
 
@@ -135,7 +137,7 @@ class TestUtils(test.TestCase):
 
         mox.ReplayAll()
 
-        inf = image_utils.qemu_img_info(TEST_PATH)
+        inf = image_utils.qemu_img_info(TEST_PATH, run_as_root=True)
 
         self.assertEqual(inf.image, 'qemu.qcow2')
         self.assertEqual(inf.backing_file, 'qemu.qcow2')
@@ -187,7 +189,7 @@ class TestUtils(test.TestCase):
 
         mox.ReplayAll()
 
-        inf = image_utils.qemu_img_info(TEST_PATH)
+        inf = image_utils.qemu_img_info(TEST_PATH, run_as_root=True)
 
         self.assertEqual(inf.image, 'qemu.qcow2')
         self.assertEqual(inf.backing_file, 'qemu.qcow2')
@@ -285,7 +287,7 @@ class TestUtils(test.TestCase):
 
         image_utils.fetch_to_raw(context, self._image_service,
                                  self.TEST_IMAGE_ID, self.TEST_DEV_PATH,
-                                 mox.IgnoreArg())
+                                 mox.IgnoreArg(), run_as_root=True)
         self._mox.VerifyAll()
 
     @mock.patch('os.stat')

cinder/tests/test_netapp_nfs.py

@@ -122,12 +122,12 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
         mox.StubOutWithMock(drv, '_get_volume_location')
         mox.StubOutWithMock(drv, 'local_path')
         mox.StubOutWithMock(drv, '_discover_file_till_timeout')
-        mox.StubOutWithMock(drv, '_set_rw_permissions_for_all')
+        mox.StubOutWithMock(drv, '_set_rw_permissions')
         drv._clone_volume(IgnoreArg(), IgnoreArg(), IgnoreArg())
         drv._get_volume_location(IgnoreArg()).AndReturn(location)
         drv.local_path(IgnoreArg()).AndReturn('/mnt')
         drv._discover_file_till_timeout(IgnoreArg()).AndReturn(True)
-        drv._set_rw_permissions_for_all(IgnoreArg())
+        drv._set_rw_permissions(IgnoreArg())
 
         mox.ReplayAll()
 
@@ -524,7 +524,7 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
         mox.StubOutWithMock(image_utils, 'qemu_img_info')
         mox.StubOutWithMock(drv, '_clone_volume')
         mox.StubOutWithMock(drv, '_discover_file_till_timeout')
-        mox.StubOutWithMock(drv, '_set_rw_permissions_for_all')
+        mox.StubOutWithMock(drv, '_set_rw_permissions')
         mox.StubOutWithMock(drv, '_resize_image_file')
         mox.StubOutWithMock(drv, '_is_share_vol_compatible')
 
@@ -532,13 +532,13 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
         drv._is_cloneable_share(IgnoreArg()).AndReturn('127.0.0.1:/share')
         drv._is_share_vol_compatible(IgnoreArg(), IgnoreArg()).AndReturn(True)
         drv._get_mount_point_for_share(IgnoreArg()).AndReturn('/mnt')
-        image_utils.qemu_img_info('/mnt/img-id').AndReturn(
+        image_utils.qemu_img_info('/mnt/img-id', run_as_root=True).\
-            self.get_img_info('raw'))
+            AndReturn(self.get_img_info('raw'))
         drv._clone_volume(
             'img-id', 'vol', share='127.0.0.1:/share', volume_id=None)
         drv._get_mount_point_for_share(IgnoreArg()).AndReturn('/mnt')
         drv._discover_file_till_timeout(IgnoreArg()).AndReturn(True)
-        drv._set_rw_permissions_for_all('/mnt/vol')
+        drv._set_rw_permissions('/mnt/vol')
         drv._resize_image_file({'name': 'vol'}, IgnoreArg())
 
         mox.ReplayAll()
@@ -556,7 +556,7 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
         mox.StubOutWithMock(image_utils, 'qemu_img_info')
         mox.StubOutWithMock(drv, '_clone_volume')
         mox.StubOutWithMock(drv, '_discover_file_till_timeout')
-        mox.StubOutWithMock(drv, '_set_rw_permissions_for_all')
+        mox.StubOutWithMock(drv, '_set_rw_permissions')
         mox.StubOutWithMock(drv, '_resize_image_file')
         mox.StubOutWithMock(image_utils, 'convert_image')
         mox.StubOutWithMock(drv, '_register_image_in_cache')
@@ -567,19 +567,20 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
             '127.0.0.1:/share')
         drv._is_share_vol_compatible(IgnoreArg(), IgnoreArg()).AndReturn(True)
         drv._get_mount_point_for_share('127.0.0.1:/share').AndReturn('/mnt')
-        image_utils.qemu_img_info('/mnt/img-id').AndReturn(
+        image_utils.qemu_img_info('/mnt/img-id', run_as_root=True).\
-            self.get_img_info('notraw'))
+            AndReturn(self.get_img_info('notraw'))
-        image_utils.convert_image(IgnoreArg(), IgnoreArg(), 'raw')
+        image_utils.convert_image(IgnoreArg(), IgnoreArg(), 'raw',
-        image_utils.qemu_img_info('/mnt/vol').AndReturn(
+                                  run_as_root=True)
-            self.get_img_info('raw'))
+        image_utils.qemu_img_info('/mnt/vol', run_as_root=True).\
+            AndReturn(self.get_img_info('raw'))
         drv._register_image_in_cache(IgnoreArg(), IgnoreArg())
         drv._get_mount_point_for_share('127.0.0.1:/share').AndReturn('/mnt')
         drv._discover_file_till_timeout(IgnoreArg()).AndReturn(True)
-        drv._set_rw_permissions_for_all('/mnt/vol')
+        drv._set_rw_permissions('/mnt/vol')
         drv._resize_image_file({'name': 'vol'}, IgnoreArg())
 
         mox.ReplayAll()
-        drv. clone_image(
+        drv.clone_image(
             volume, ('nfs://127.0.0.1/share/img-id', None), 'image_id', {})
         mox.VerifyAll()
 
@@ -605,11 +606,12 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
             '127.0.0.1:/share')
         drv._is_share_vol_compatible(IgnoreArg(), IgnoreArg()).AndReturn(True)
         drv._get_mount_point_for_share('127.0.0.1:/share').AndReturn('/mnt')
-        image_utils.qemu_img_info('/mnt/img-id').AndReturn(
+        image_utils.qemu_img_info('/mnt/img-id', run_as_root=True).\
-            self.get_img_info('notraw'))
+            AndReturn(self.get_img_info('notraw'))
-        image_utils.convert_image(IgnoreArg(), IgnoreArg(), 'raw')
+        image_utils.convert_image(IgnoreArg(), IgnoreArg(), 'raw',
-        image_utils.qemu_img_info('/mnt/vol').AndReturn(
+                                  run_as_root=True)
-            self.get_img_info('raw'))
+        image_utils.qemu_img_info('/mnt/vol', run_as_root=True).\
+            AndReturn(self.get_img_info('raw'))
         drv._register_image_in_cache(IgnoreArg(), IgnoreArg())
         drv.local_path(IgnoreArg()).AndReturn('/mnt/vol')
         drv._discover_file_till_timeout(IgnoreArg()).AndReturn(False)
@@ -635,7 +637,7 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
         mox.StubOutWithMock(image_utils, 'qemu_img_info')
         mox.StubOutWithMock(drv, '_clone_volume')
         mox.StubOutWithMock(drv, '_discover_file_till_timeout')
-        mox.StubOutWithMock(drv, '_set_rw_permissions_for_all')
+        mox.StubOutWithMock(drv, '_set_rw_permissions')
         mox.StubOutWithMock(drv, '_resize_image_file')
         mox.StubOutWithMock(image_utils, 'convert_image')
         mox.StubOutWithMock(drv, '_register_image_in_cache')
@@ -649,15 +651,16 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
             '127.0.0.1:/share')
         drv._is_share_vol_compatible(IgnoreArg(), IgnoreArg()).AndReturn(True)
         drv._get_mount_point_for_share('127.0.0.1:/share').AndReturn('/mnt')
-        image_utils.qemu_img_info('/mnt/img-id').AndReturn(
+        image_utils.qemu_img_info('/mnt/img-id', run_as_root=True).\
-            self.get_img_info('notraw'))
+            AndReturn(self.get_img_info('notraw'))
-        image_utils.convert_image(IgnoreArg(), IgnoreArg(), 'raw')
+        image_utils.convert_image(IgnoreArg(), IgnoreArg(), 'raw',
-        image_utils.qemu_img_info('/mnt/vol').AndReturn(
+                                  run_as_root=True)
-            self.get_img_info('raw'))
+        image_utils.qemu_img_info('/mnt/vol', run_as_root=True).\
+            AndReturn(self.get_img_info('raw'))
         drv._register_image_in_cache(IgnoreArg(), IgnoreArg())
         drv.local_path(IgnoreArg()).AndReturn('/mnt/vol')
         drv._discover_file_till_timeout(IgnoreArg()).AndReturn(True)
-        drv._set_rw_permissions_for_all('/mnt/vol')
+        drv._set_rw_permissions('/mnt/vol')
         drv._resize_image_file(
             IgnoreArg(), IgnoreArg()).AndRaise(exception.InvalidResults())
         drv.local_path(IgnoreArg()).AndReturn('/mnt/vol')
@@ -1079,7 +1082,7 @@ class NetappDirectCmodeNfsDriverOnlyTestCase(test.TestCase):
         drv._execute.assert_called_once_with('cof_path', 'ip1', 'ip1',
                                              '/openstack/img-cache-imgid',
                                              '/exp_path/name',
-                                             run_as_root=False,
+                                             run_as_root=True,
                                              check_exit_code=0)
         drv._post_clone_image.assert_called_with(volume)
         drv._get_provider_location.assert_called_with('vol_id')

cinder/tests/test_nfs.py

@@ -46,11 +46,19 @@ class DumbVolume(object):
 
 class RemoteFsDriverTestCase(test.TestCase):
     TEST_FILE_NAME = 'test.txt'
+    TEST_EXPORT = 'nas-host1:/export'
+    TEST_MNT_POINT = '/mnt/nas'
 
     def setUp(self):
         super(RemoteFsDriverTestCase, self).setUp()
         self._driver = remotefs.RemoteFSDriver()
         self._mox = mox_lib.Mox()
+        self.configuration = mox_lib.MockObject(conf.Configuration)
+        self.configuration.append_config_values(mox_lib.IgnoreArg())
+        self.configuration.nas_secure_file_permissions = 'false'
+        self.configuration.nas_secure_file_operations = 'false'
+        self._driver = remotefs.RemoteFSDriver(
+            configuration=self.configuration)
         self.addCleanup(self._mox.UnsetStubs)
 
     def test_create_sparsed_file(self):
@@ -107,6 +115,229 @@ class RemoteFsDriverTestCase(test.TestCase):
 
         mox.VerifyAll()
 
+    @mock.patch.object(remotefs, 'LOG')
+    def test_set_rw_permissions_with_secure_file_permissions(self, LOG):
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_EXPORT]
+        drv.configuration.nas_secure_file_permissions = 'true'
+        self.stubs.Set(drv, '_execute', mock.Mock())
+
+        drv._set_rw_permissions(self.TEST_FILE_NAME)
+
+        self.assertFalse(LOG.warn.called)
+
+    @mock.patch.object(remotefs, 'LOG')
+    def test_set_rw_permissions_without_secure_file_permissions(self, LOG):
+        drv = self._driver
+        self.configuration.nas_secure_file_permissions = 'false'
+        self.stubs.Set(drv, '_execute', mock.Mock())
+
+        drv._set_rw_permissions(self.TEST_FILE_NAME)
+
+        self.assertTrue(LOG.warn.called)
+        warn_msg = "%s is being set with open permissions: ugo+rw" % \
+                   self.TEST_FILE_NAME
+        LOG.warn.assert_called_once_with(warn_msg)
+
+    @mock.patch('os.path.join')
+    @mock.patch('os.path.isfile', return_value=False)
+    def test_determine_nas_security_options_when_auto_and_new_install(
+            self,
+            mock_isfile,
+            mock_join):
+        """Test the setting of the NAS Security Option
+
+         In this test case, we will create the marker file. No pre-exxisting
+         Cinder volumes found during bootup.
+         """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_EXPORT]
+        file_path = '%s/.cinderSecureEnvIndicator' % self.TEST_MNT_POINT
+        is_new_install = True
+
+        drv._ensure_shares_mounted = mock.Mock()
+        nas_mount = drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+        mock_join.return_value = file_path
+
+        secure_file_permissions = 'auto'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_permissions,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'true')
+
+        secure_file_operations = 'auto'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_operations,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'true')
+
+    @mock.patch('os.path.join')
+    @mock.patch('os.path.isfile')
+    def test_determine_nas_security_options_when_auto_and_new_install_exists(
+            self,
+            isfile,
+            join):
+        """Test the setting of the NAS Security Option
+
+        In this test case, the marker file already exists. Cinder volumes
+        found during bootup.
+        """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_EXPORT]
+        file_path = '%s/.cinderSecureEnvIndicator' % self.TEST_MNT_POINT
+        is_new_install = False
+
+        drv._ensure_shares_mounted = mock.Mock()
+        nas_mount = drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+        join.return_value = file_path
+        isfile.return_value = True
+
+        secure_file_permissions = 'auto'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_permissions,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'true')
+
+        secure_file_operations = 'auto'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_operations,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'true')
+
+    @mock.patch('os.path.join')
+    @mock.patch('os.path.isfile')
+    def test_determine_nas_security_options_when_auto_and_old_install(self,
+                                                                      isfile,
+                                                                      join):
+        """Test the setting of the NAS Security Option
+
+        In this test case, the marker file does not exist. There are also
+        pre-existing Cinder volumes.
+        """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_EXPORT]
+        file_path = '%s/.cinderSecureEnvIndicator' % self.TEST_MNT_POINT
+        is_new_install = False
+
+        drv._ensure_shares_mounted = mock.Mock()
+        nas_mount = drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+        join.return_value = file_path
+        isfile.return_value = False
+
+        secure_file_permissions = 'auto'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_permissions,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'false')
+
+        secure_file_operations = 'auto'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_operations,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'false')
+
+    def test_determine_nas_security_options_when_admin_set_true(self):
+        """Test the setting of the NAS Security Option
+
+        In this test case, the Admin set the flag to 'true'.
+        """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_EXPORT]
+        is_new_install = False
+
+        drv._ensure_shares_mounted = mock.Mock()
+        nas_mount = drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+
+        secure_file_permissions = 'true'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_permissions,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'true')
+
+        secure_file_operations = 'true'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_operations,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'true')
+
+    def test_determine_nas_security_options_when_admin_set_false(self):
+        """Test the setting of the NAS Security Option
+
+        In this test case, the Admin set the flag to 'true'.
+        """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_EXPORT]
+        is_new_install = False
+
+        drv._ensure_shares_mounted = mock.Mock()
+        nas_mount = drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+
+        secure_file_permissions = 'false'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_permissions,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'false')
+
+        secure_file_operations = 'false'
+        nas_option = drv._determine_nas_security_option_setting(
+            secure_file_operations,
+            nas_mount, is_new_install)
+
+        self.assertEqual(nas_option, 'false')
+
+    @mock.patch.object(remotefs, 'LOG')
+    def test_set_nas_security_options(self, LOG):
+        """Test setting of NAS Security options.
+
+        The RemoteFS driver will force set options to false. The derived
+        objects will provide an inherited interface to properly set options.
+        """
+        drv = self._driver
+        is_new_install = False
+
+        drv.set_nas_security_options(is_new_install)
+
+        self.assertEqual(drv.configuration.nas_secure_file_operations, 'false')
+        self.assertEqual(drv.configuration.nas_secure_file_permissions,
+                         'false')
+        self.assertTrue(LOG.warn.called)
+
+    def test_secure_file_operations_enabled_true(self):
+        """Test nas_secure_file_operations = 'true'
+
+        Networked file system based drivers may support secure file
+        operations. This test verifies the settings when secure.
+        """
+        drv = self._driver
+        self.configuration.nas_secure_file_operations = 'true'
+        ret_flag = drv.secure_file_operations_enabled()
+        self.assertTrue(ret_flag)
+
+    def test_secure_file_operations_enabled_false(self):
+        """Test nas_secure_file_operations = 'false'
+
+        Networked file system based drivers may support secure file
+        operations. This test verifies the settings when not secure.
+        """
+        drv = self._driver
+        self.configuration.nas_secure_file_operations = 'false'
+        ret_flag = drv.secure_file_operations_enabled()
+        self.assertFalse(ret_flag)
+
 
 class NfsDriverTestCase(test.TestCase):
     """Test case for NFS driver."""
@@ -135,6 +366,8 @@ class NfsDriverTestCase(test.TestCase):
         self.configuration.nfs_oversub_ratio = 1.0
         self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE
         self.configuration.nfs_mount_options = None
+        self.configuration.nas_secure_file_permissions = 'false'
+        self.configuration.nas_secure_file_operations = 'false'
         self.configuration.volume_dd_blocksize = '1M'
         self._driver = nfs.NfsDriver(configuration=self.configuration)
         self._driver.shares = {}
@@ -176,15 +409,18 @@ class NfsDriverTestCase(test.TestCase):
         mox.StubOutWithMock(image_utils, 'fetch_to_raw')
         image_utils.fetch_to_raw(None, None, None, TEST_IMG_SOURCE,
                                  mox_lib.IgnoreArg(),
-                                 size=self.TEST_SIZE_IN_GB)
+                                 size=self.TEST_SIZE_IN_GB,
+                                 run_as_root=True)
 
         mox.StubOutWithMock(image_utils, 'resize_image')
-        image_utils.resize_image(TEST_IMG_SOURCE, self.TEST_SIZE_IN_GB)
+        image_utils.resize_image(TEST_IMG_SOURCE, self.TEST_SIZE_IN_GB,
+                                 run_as_root=True)
 
         mox.StubOutWithMock(image_utils, 'qemu_img_info')
         data = mox_lib.MockAnything()
         data.virtual_size = 1 * units.Gi
-        image_utils.qemu_img_info(TEST_IMG_SOURCE).AndReturn(data)
+        image_utils.qemu_img_info(TEST_IMG_SOURCE,
+                                  run_as_root=True).AndReturn(data)
 
         mox.ReplayAll()
 
@@ -382,11 +618,10 @@ class NfsDriverTestCase(test.TestCase):
         mox = self._mox
         drv = self._driver
         self.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE
-
         mox.StubOutWithMock(os.path, 'exists')
         os.path.exists(self.TEST_SHARES_CONFIG_FILE).AndReturn(True)
         mox.StubOutWithMock(drv, '_execute')
-        drv._execute('mount.nfs', check_exit_code=False, run_as_root=True).\
+        drv._execute('mount.nfs', check_exit_code=False, run_as_root=False).\
             AndRaise(OSError(errno.ENOENT, 'No such file or directory'))
 
         mox.ReplayAll()
@@ -470,10 +705,10 @@ class NfsDriverTestCase(test.TestCase):
         cfg.CONF.set_override('nfs_sparsed_volumes', True)
 
         mox.StubOutWithMock(drv, '_create_sparsed_file')
-        mox.StubOutWithMock(drv, '_set_rw_permissions_for_all')
+        mox.StubOutWithMock(drv, '_set_rw_permissions')
 
         drv._create_sparsed_file(IgnoreArg(), IgnoreArg())
-        drv._set_rw_permissions_for_all(IgnoreArg())
+        drv._set_rw_permissions(IgnoreArg())
 
         mox.ReplayAll()
 
@@ -490,10 +725,10 @@ class NfsDriverTestCase(test.TestCase):
         cfg.CONF.set_override('nfs_sparsed_volumes', False)
 
         mox.StubOutWithMock(drv, '_create_regular_file')
-        mox.StubOutWithMock(drv, '_set_rw_permissions_for_all')
+        mox.StubOutWithMock(drv, '_set_rw_permissions')
 
         drv._create_regular_file(IgnoreArg(), IgnoreArg())
-        drv._set_rw_permissions_for_all(IgnoreArg())
+        drv._set_rw_permissions(IgnoreArg())
 
         mox.ReplayAll()
 
@@ -702,7 +937,8 @@ class NfsDriverTestCase(test.TestCase):
                                            return_value=True):
                         drv.extend_volume(volume, newSize)
 
-                        resize.assert_called_once_with(path, newSize)
+                        resize.assert_called_once_with(path, newSize,
+                                                       run_as_root=True)
 
     def test_extend_volume_failure(self):
         """Error during extend operation."""
@@ -757,3 +993,50 @@ class NfsDriverTestCase(test.TestCase):
         with mock.patch.object(image_utils, 'qemu_img_info',
                                return_value=data):
             self.assertFalse(drv._is_file_size_equal(path, size))
+
+    @mock.patch.object(nfs, 'LOG')
+    def test_set_nas_security_options_when_true(self, LOG):
+        """Test higher level setting of NAS Security options.
+
+        The NFS driver overrides the base method with a driver specific
+        version.
+        """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_NFS_EXPORT1]
+        is_new_install = True
+
+        drv._ensure_shares_mounted = mock.Mock()
+        drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+        drv._determine_nas_security_option_setting = mock.Mock(
+            return_value='true')
+
+        drv.set_nas_security_options(is_new_install)
+
+        self.assertEqual(drv.configuration.nas_secure_file_operations, 'true')
+        self.assertEqual(drv.configuration.nas_secure_file_permissions, 'true')
+        self.assertFalse(LOG.warn.called)
+
+    @mock.patch.object(nfs, 'LOG')
+    def test_set_nas_security_options_when_false(self, LOG):
+        """Test higher level setting of NAS Security options.
+
+        The NFS driver overrides the base method with a driver specific
+        version.
+        """
+        drv = self._driver
+        drv._mounted_shares = [self.TEST_NFS_EXPORT1]
+        is_new_install = False
+
+        drv._ensure_shares_mounted = mock.Mock()
+        drv._get_mount_point_for_share = mock.Mock(
+            return_value=self.TEST_MNT_POINT)
+        drv._determine_nas_security_option_setting = mock.Mock(
+            return_value='false')
+
+        drv.set_nas_security_options(is_new_install)
+
+        self.assertEqual(drv.configuration.nas_secure_file_operations, 'false')
+        self.assertEqual(drv.configuration.nas_secure_file_permissions,
+                         'false')
+        self.assertTrue(LOG.warn.called)

cinder/tests/test_remotefs.py

@@ -225,7 +225,7 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
             self._FAKE_VOLUME_NAME)
 
         self._driver._execute = mock.Mock()
-        self._driver._set_rw_permissions_for_all = mock.Mock()
+        self._driver._set_rw_permissions = mock.Mock()
         self._driver._qemu_img_info = mock.Mock(
             return_value=mock.Mock(file_format=mock.sentinel.backing_fmt))
 

cinder/tests/test_smbfs.py

@@ -319,6 +319,7 @@ class SmbFsTestCase(test.TestCase):
             return_value=not extend_failed)
         drv._qemu_img_info = mock.Mock(
             return_value=mock.Mock(file_format=image_format))
+        drv._delete = mock.Mock()
 
         with contextlib.nested(
                 mock.patch.object(image_utils, 'resize_image'),

cinder/tests/test_volume.py

@@ -3270,6 +3270,16 @@ class VolumeTestCase(BaseVolumeTestCase):
         # Group is not deleted
         self.assertEqual(cg['status'], 'available')
 
+    def test_secure_file_operations_enabled(self):
+        """Test secure file operations setting for base driver.
+
+        General, non network file system based drivers do not have
+        anything to do with "secure_file_operations". This test verifies that
+        calling the method always returns False.
+        """
+        ret_flag = self.volume.driver.secure_file_operations_enabled()
+        self.assertFalse(ret_flag)
+
 
 class CopyVolumeToImageTestCase(BaseVolumeTestCase):
     def fake_local_path(self, volume):

cinder/volume/driver.py

@@ -506,7 +506,10 @@ class VolumeDriver(object):
         device = connector.connect_volume(conn['data'])
         host_device = device['path']
 
-        if not connector.check_valid_device(host_device):
+        # Secure network file systems will NOT run as root.
+        root_access = not self.secure_file_operations_enabled()
+
+        if not connector.check_valid_device(host_device, root_access):
             raise exception.DeviceUnavailable(path=host_device,
                                               reason=(_("Unable to access "
                                                         "the backend storage "
@@ -548,6 +551,12 @@ class VolumeDriver(object):
 
         try:
             volume_path = attach_info['device']['path']
+
+            # Secure network file systems will not chown files.
+            if self.secure_file_operations_enabled():
+                with fileutils.file_open(volume_path) as volume_file:
+                    backup_service.backup(backup, volume_file)
+            else:
                 with utils.temporary_chown(volume_path):
                     with fileutils.file_open(volume_path) as volume_file:
                         backup_service.backup(backup, volume_file)
@@ -567,9 +576,16 @@ class VolumeDriver(object):
 
         try:
             volume_path = attach_info['device']['path']
-            with utils.temporary_chown(volume_path):
+
+            # Secure network file systems will not chown files.
+            if self.secure_file_operations_enabled():
                 with fileutils.file_open(volume_path, 'wb') as volume_file:
                     backup_service.restore(backup, volume['id'], volume_file)
+            else:
+                with utils.temporary_chown(volume_path):
+                    with fileutils.file_open(volume_path, 'wb') as volume_file:
+                        backup_service.restore(backup, volume['id'],
+                                               volume_file)
 
         finally:
             self._detach_volume(context, attach_info, volume, properties)
@@ -828,6 +844,15 @@ class VolumeDriver(object):
         """
         return None
 
+    def secure_file_operations_enabled(self):
+        """Determine if driver is running in Secure File Operations mode.
+
+        The Cinder Volume driver needs to query if this driver is running
+        in a secure file operations mode. By default, it is False: any driver
+        that does support secure file operations should override this method.
+        """
+        return False
+
 
 class ISCSIDriver(VolumeDriver):
     """Executes commands relating to ISCSI volumes.

cinder/volume/drivers/netapp/common.py

@@ -30,9 +30,9 @@ from cinder.volume.drivers.netapp.options import netapp_proxy_opts
 LOG = logging.getLogger(__name__)
 
 
-#NOTE(singn): Holds family:{protocol:driver} registration information.
+# NOTE(singn): Holds family:{protocol:driver} registration information.
-#Plug in new families and protocols to support new drivers.
+# Plug in new families and protocols to support new drivers.
-#No other code modification required.
+# No other code modification required.
 netapp_unified_plugin_registry =\
     {'ontap_cluster':
      {
@@ -54,9 +54,9 @@ netapp_unified_plugin_registry =\
      },
      }
 
-#NOTE(singn): Holds family:protocol information.
+# NOTE(singn): Holds family:protocol information.
-#Protocol represents the default protocol driver option
+# Protocol represents the default protocol driver option
-#in case no protocol is specified by the user in configuration.
+# in case no protocol is specified by the user in configuration.
 netapp_family_default =\
     {
         'ontap_cluster': 'nfs',

cinder/volume/drivers/netapp/nfs.py

@@ -99,9 +99,10 @@ class NetAppNFSDriver(nfs.NfsDriver):
         share = self._get_volume_location(snapshot.volume_id)
         volume['provider_location'] = share
         path = self.local_path(volume)
+        run_as_root = self._execute_as_root
 
         if self._discover_file_till_timeout(path):
-            self._set_rw_permissions_for_all(path)
+            self._set_rw_permissions(path)
             if vol_size != snap_size:
                 try:
                     self.extend_volume(volume, vol_size)
@@ -110,7 +111,7 @@ class NetAppNFSDriver(nfs.NfsDriver):
                         LOG.error(
                             _("Resizing %s failed. Cleaning volume."),
                             volume.name)
-                        self._execute('rm', path, run_as_root=True)
+                        self._execute('rm', path, run_as_root=run_as_root)
         else:
             raise exception.CinderException(
                 _("NFS file %s not discovered.") % volume['name'])
@@ -131,7 +132,7 @@ class NetAppNFSDriver(nfs.NfsDriver):
             return True
 
         self._execute('rm', self._get_volume_path(nfs_mount, snapshot.name),
-                      run_as_root=True)
+                      run_as_root=self._execute_as_root)
 
     def _get_client(self):
         """Creates client for server."""
@@ -209,14 +210,15 @@ class NetAppNFSDriver(nfs.NfsDriver):
         path = self.local_path(volume)
 
         if self._discover_file_till_timeout(path):
-            self._set_rw_permissions_for_all(path)
+            self._set_rw_permissions(path)
             if vol_size != src_vol_size:
                 try:
                     self.extend_volume(volume, vol_size)
                 except Exception as e:
                     LOG.error(
                         _("Resizing %s failed. Cleaning volume."), volume.name)
-                    self._execute('rm', path, run_as_root=True)
+                    self._execute('rm', path,
+                                  run_as_root=self._execute_as_root)
                     raise e
         else:
             raise exception.CinderException(
@@ -283,7 +285,7 @@ class NetAppNFSDriver(nfs.NfsDriver):
             LOG.debug('Image cache cleaning in progress. Returning... ')
             return
         else:
-            #set cleaning to True
+            # Set cleaning to True
             self.cleaning = True
             t = Timer(0, self._clean_image_cache)
             t.start()
@@ -333,7 +335,8 @@ class NetAppNFSDriver(nfs.NfsDriver):
         threshold_minutes = self.configuration.expiry_thres_minutes
         cmd = ['find', mount_fs, '-maxdepth', '1', '-name',
                'img-cache*', '-amin', '+%s' % (threshold_minutes)]
-        res, __ = self._execute(*cmd, run_as_root=True)
+        res, _err = self._execute(*cmd,
+                                  run_as_root=self._execute_as_root)
         if res:
             old_file_paths = res.strip('\n').split('\n')
             mount_fs_len = len(mount_fs)
@@ -369,7 +372,7 @@ class NetAppNFSDriver(nfs.NfsDriver):
         try:
             LOG.debug('Deleting file at path %s', path)
             cmd = ['rm', '-f', path]
-            self._execute(*cmd, run_as_root=True)
+            self._execute(*cmd, run_as_root=self._execute_as_root)
             return True
         except Exception as ex:
             LOG.warning(_('Exception during deleting %s'), ex.__str__())
@@ -444,13 +447,16 @@ class NetAppNFSDriver(nfs.NfsDriver):
         cloned = False
         image_location = self._construct_image_nfs_url(image_location)
         share = self._is_cloneable_share(image_location)
+        run_as_root = self._execute_as_root
+
         if share and self._is_share_vol_compatible(volume, share):
             LOG.debug('Share is cloneable %s', share)
             volume['provider_location'] = share
             (__, ___, img_file) = image_location.rpartition('/')
             dir_path = self._get_mount_point_for_share(share)
             img_path = '%s/%s' % (dir_path, img_file)
-            img_info = image_utils.qemu_img_info(img_path)
+            img_info = image_utils.qemu_img_info(img_path,
+                                                 run_as_root=run_as_root)
             if img_info.file_format == 'raw':
                 LOG.debug('Image is raw %s', image_id)
                 self._clone_volume(
@@ -462,8 +468,9 @@ class NetAppNFSDriver(nfs.NfsDriver):
                     _('Image will locally be converted to raw %s'),
                     image_id)
                 dst = '%s/%s' % (dir_path, volume['name'])
-                image_utils.convert_image(img_path, dst, 'raw')
+                image_utils.convert_image(img_path, dst, 'raw',
-                data = image_utils.qemu_img_info(dst)
+                                          run_as_root=run_as_root)
+                data = image_utils.qemu_img_info(dst, run_as_root=run_as_root)
                 if data.file_format != "raw":
                     raise exception.InvalidResults(
                         _("Converted to raw, but"
@@ -479,7 +486,7 @@ class NetAppNFSDriver(nfs.NfsDriver):
         LOG.info(_('Performing post clone for %s'), volume['name'])
         vol_path = self.local_path(volume)
         if self._discover_file_till_timeout(vol_path):
-            self._set_rw_permissions_for_all(vol_path)
+            self._set_rw_permissions(vol_path)
             self._resize_image_file(vol_path, volume['size'])
             return True
         raise exception.InvalidResults(
@@ -492,7 +499,8 @@ class NetAppNFSDriver(nfs.NfsDriver):
             return
         else:
             LOG.info(_('Resizing file to %sG'), new_size)
-            image_utils.resize_image(path, new_size)
+            image_utils.resize_image(path, new_size,
+                                     run_as_root=self._execute_as_root)
             if self._is_file_size_equal(path, new_size):
                 return
             else:
@@ -501,7 +509,8 @@ class NetAppNFSDriver(nfs.NfsDriver):
 
     def _is_file_size_equal(self, path, size):
         """Checks if file size at path is equal to size."""
-        data = image_utils.qemu_img_info(path)
+        data = image_utils.qemu_img_info(path,
+                                         run_as_root=self._execute_as_root)
         virt_size = data.virtual_size / units.Gi
         if virt_size == size:
             return True
@@ -639,7 +648,8 @@ class NetAppNFSDriver(nfs.NfsDriver):
             if os.path.exists(dst):
                 LOG.warn(_("Destination %s already exists."), dst)
                 return False
-            self._execute('mv', src, dst, run_as_root=True)
+            self._execute('mv', src, dst,
+                          run_as_root=self._execute_as_root)
             return True
 
         try:
@@ -1218,7 +1228,8 @@ class NetAppDirectCmodeNfsDriver (NetAppDirectNfsDriver):
                     dst_path = os.path.join(
                         self._get_export_path(volume['id']), volume['name'])
                     self._execute(col_path, src_ip, dst_ip,
-                                  src_path, dst_path, run_as_root=False,
+                                  src_path, dst_path,
+                                  run_as_root=self._execute_as_root,
                                   check_exit_code=0)
                     self._register_image_in_cache(volume, image_id)
                     LOG.debug("Copied image from cache to volume %s using"
@@ -1264,6 +1275,7 @@ class NetAppDirectCmodeNfsDriver (NetAppDirectNfsDriver):
         img_info = image_service.show(context, image_id)
         dst_share = self._get_provider_location(volume['id'])
         self._check_share_can_hold_size(dst_share, img_info['size'])
+        run_as_root = self._execute_as_root
 
         dst_dir = self._get_mount_point_for_share(dst_share)
         dst_img_local = os.path.join(dst_dir, tmp_img_file)
@@ -1274,7 +1286,7 @@ class NetAppDirectCmodeNfsDriver (NetAppDirectNfsDriver):
                 dst_img_serv_path = os.path.join(
                     self._get_export_path(volume['id']), tmp_img_file)
                 self._execute(col_path, src_ip, dst_ip, src_path,
-                              dst_img_serv_path, run_as_root=False,
+                              dst_img_serv_path, run_as_root=run_as_root,
                               check_exit_code=0)
             else:
                 self._clone_file_dst_exists(dst_share, img_file, tmp_img_file)
@@ -1299,8 +1311,10 @@ class NetAppDirectCmodeNfsDriver (NetAppDirectNfsDriver):
                 self._check_share_can_hold_size(dst_share, img_info['size'])
                 try:
                     image_utils.convert_image(dst_img_local,
-                                              dst_img_conv_local, 'raw')
+                                              dst_img_conv_local, 'raw',
-                    data = image_utils.qemu_img_info(dst_img_conv_local)
+                                              run_as_root=run_as_root)
+                    data = image_utils.qemu_img_info(dst_img_conv_local,
+                                                     run_as_root=run_as_root)
                     if data.file_format != "raw":
                         raise exception.InvalidResults(
                             _("Converted to raw, but format is now %s.")

cinder/volume/drivers/nfs.py

@@ -28,11 +28,11 @@ from cinder.openstack.common import units
 from cinder import utils
 from cinder.volume.drivers import remotefs
 
-VERSION = '1.1.0'
+VERSION = '1.2.0'
 
 LOG = logging.getLogger(__name__)
 
-volume_opts = [
+nfs_opts = [
     cfg.StrOpt('nfs_shares_config',
                default='/etc/cinder/nfs_shares',
                help='File with the list of available nfs shares'),
@@ -61,7 +61,7 @@ volume_opts = [
 ]
 
 CONF = cfg.CONF
-CONF.register_opts(volume_opts)
+CONF.register_opts(nfs_opts)
 
 
 class NfsDriver(remotefs.RemoteFSDriver):
@@ -77,7 +77,7 @@ class NfsDriver(remotefs.RemoteFSDriver):
     def __init__(self, execute=putils.execute, *args, **kwargs):
         self._remotefsclient = None
         super(NfsDriver, self).__init__(*args, **kwargs)
-        self.configuration.append_config_values(volume_opts)
+        self.configuration.append_config_values(nfs_opts)
         root_helper = utils.get_root_helper()
         # base bound to instance is used in RemoteFsConnector.
         self.base = getattr(self.configuration,
@@ -127,15 +127,23 @@ class NfsDriver(remotefs.RemoteFSDriver):
 
         self.shares = {}  # address : options
 
-        # Check if mount.nfs is installed
+        # Check if mount.nfs is installed on this system; note that we don't
+        # need to be root to see if the package is installed.
+        package = 'mount.nfs'
         try:
-            self._execute('mount.nfs', check_exit_code=False, run_as_root=True)
+            self._execute(package, check_exit_code=False,
+                          run_as_root=False)
         except OSError as exc:
             if exc.errno == errno.ENOENT:
-                raise exception.NfsException('mount.nfs is not installed')
+                msg = _('%s is not installed') % package
+                raise exception.NfsException(msg)
             else:
                 raise exc
 
+        # Now that all configuration data has been loaded (shares),
+        # we can "set" our final NAS file security options.
+        self.set_nas_security_options(self._is_voldb_empty_at_startup)
+
     def _ensure_share_mounted(self, nfs_share):
         mnt_flags = []
         if self.shares.get(nfs_share) is not None:
@@ -227,17 +235,19 @@ class NfsDriver(remotefs.RemoteFSDriver):
 
         :param nfs_share: example 172.18.194.100:/var/nfs
         """
+        run_as_root = self._execute_as_root
 
         mount_point = self._get_mount_point_for_share(nfs_share)
 
         df, _ = self._execute('stat', '-f', '-c', '%S %b %a', mount_point,
-                              run_as_root=True)
+                              run_as_root=run_as_root)
         block_size, blocks_total, blocks_avail = map(float, df.split())
         total_available = block_size * blocks_avail
         total_size = block_size * blocks_total
 
         du, _ = self._execute('du', '-sb', '--apparent-size', '--exclude',
-                              '*snapshot*', mount_point, run_as_root=True)
+                              '*snapshot*', mount_point,
+                              run_as_root=run_as_root)
         total_allocated = float(du.split()[0])
         return total_size, total_available, total_allocated
 
@@ -255,13 +265,69 @@ class NfsDriver(remotefs.RemoteFSDriver):
                                               % (volume['id'], new_size))
         path = self.local_path(volume)
         LOG.info(_('Resizing file to %sG...'), new_size)
-        image_utils.resize_image(path, new_size)
+        image_utils.resize_image(path, new_size,
+                                 run_as_root=self._execute_as_root)
         if not self._is_file_size_equal(path, new_size):
             raise exception.ExtendVolumeError(
                 reason='Resizing image file failed.')
 
     def _is_file_size_equal(self, path, size):
         """Checks if file size at path is equal to size."""
-        data = image_utils.qemu_img_info(path)
+        data = image_utils.qemu_img_info(path,
+                                         run_as_root=self._execute_as_root)
         virt_size = data.virtual_size / units.Gi
         return virt_size == size
+
+    def set_nas_security_options(self, is_new_cinder_install):
+        """Determine the setting to use for Secure NAS options.
+
+        Value of each NAS Security option is checked and updated. If the
+        option is currently 'auto', then it is set to either true or false
+        based upon if this is a new Cinder installation. The RemoteFS variable
+        '_execute_as_root' will be updated for this driver.
+
+        :param is_new_cinder_install: bool indication of new Cinder install
+        """
+        doc_html = "http://docs.openstack.org/admin-guide-cloud/content" \
+                   "/nfs_backend.html"
+
+        self._ensure_shares_mounted()
+        if not self._mounted_shares:
+            raise exception.NfsNoSharesMounted()
+
+        nfs_mount = self._get_mount_point_for_share(self._mounted_shares[0])
+
+        self.configuration.nas_secure_file_permissions = \
+            self._determine_nas_security_option_setting(
+                self.configuration.nas_secure_file_permissions,
+                nfs_mount, is_new_cinder_install)
+
+        LOG.debug('NAS variable secure_file_permissions setting is: %s' %
+                  self.configuration.nas_secure_file_permissions)
+
+        if self.configuration.nas_secure_file_permissions == 'false':
+            LOG.warn(_("The NAS file permissions mode will be 666 (allowing "
+                       "other/world read & write access). This is considered "
+                       "an insecure NAS environment. Please see %s for "
+                       "information on a secure NFS configuration.") %
+                     doc_html)
+
+        self.configuration.nas_secure_file_operations = \
+            self._determine_nas_security_option_setting(
+                self.configuration.nas_secure_file_operations,
+                nfs_mount, is_new_cinder_install)
+
+        # If secure NAS, update the '_execute_as_root' flag to not
+        # run as the root user; run as process' user ID.
+        if self.configuration.nas_secure_file_operations == 'true':
+            self._execute_as_root = False
+
+        LOG.debug('NAS variable secure_file_operations setting is: %s' %
+                  self.configuration.nas_secure_file_operations)
+
+        if self.configuration.nas_secure_file_operations == 'false':
+            LOG.warn(_("The NAS file operations will be run as root: allowing "
+                       "root level access at the storage backend. This is "
+                       "considered an insecure NAS environment. Please see %s "
+                       "for information on a secure NAS configuration.") %
+                     doc_html)

cinder/volume/drivers/remotefs.py

@@ -49,6 +49,25 @@ nas_opts = [
     cfg.StrOpt('nas_private_key',
                default='',
                help='Filename of private key to use for SSH authentication.'),
+    cfg.StrOpt('nas_secure_file_operations',
+               default='auto',
+               help=('Allow network-attached storage systems to operate in a '
+                     'secure environment where root level access is not '
+                     'permitted. If set to False, access is as the root user '
+                     'and insecure. If set to True, access is not as root. '
+                     'If set to auto, a check is done to determine if this is '
+                     'a new installation: True is used if so, otherwise '
+                     'False. Default is auto.')),
+    cfg.StrOpt('nas_secure_file_permissions',
+               default='auto',
+               help=('Set more secure file permissions on network-attached '
+                     'storage volume files to restrict broad other/world '
+                     'access. If set to False, volumes are created with open '
+                     'permissions. If set to True, volumes are created with '
+                     'permissions for the cinder user and group (660). If '
+                     'set to auto, a check is done to determine if '
+                     'this is a new installation: True is used if so, '
+                     'otherwise False. Default is auto.')),
 ]
 
 CONF = cfg.CONF
@@ -67,6 +86,11 @@ class RemoteFSDriver(driver.VolumeDriver):
         super(RemoteFSDriver, self).__init__(*args, **kwargs)
         self.shares = {}
         self._mounted_shares = []
+        self._execute_as_root = True
+        self._is_voldb_empty_at_startup = kwargs.pop('is_vol_db_empty', None)
+
+        if self.configuration:
+            self.configuration.append_config_values(nas_opts)
 
     def check_for_setup_error(self):
         """Just to override parent behavior."""
@@ -88,6 +112,28 @@ class RemoteFSDriver(driver.VolumeDriver):
             'mount_point_base': self._get_mount_point_base()
         }
 
+    def do_setup(self, context):
+        """Any initialization the volume driver does while starting."""
+        super(RemoteFSDriver, self).do_setup(context)
+
+        # Validate the settings for our secure file options.
+        self.configuration.nas_secure_file_permissions = \
+            self.configuration.nas_secure_file_permissions.lower()
+        self.configuration.nas_secure_file_operations = \
+            self.configuration.nas_secure_file_operations.lower()
+        valid_secure_opts = ['auto', 'true', 'false']
+        secure_options = {'nas_secure_file_permissions':
+                          self.configuration.nas_secure_file_permissions,
+                          'nas_secure_file_operations':
+                          self.configuration.nas_secure_file_operations}
+        for opt_name, opt_value in secure_options.iteritems():
+            if opt_value not in valid_secure_opts:
+                err_parms = {'name': opt_name, 'value': opt_value}
+                msg = _("NAS config '%(name)s=%(value)s' invalid. Must be "
+                        "'auto', 'true', or 'false'") % err_parms
+                LOG.error(msg)
+                raise exception.InvalidConfigurationValue(msg)
+
     def _get_mount_point_base(self):
         """Returns the mount point base for the remote fs.
 
@@ -132,7 +178,7 @@ class RemoteFSDriver(driver.VolumeDriver):
         else:
             self._create_regular_file(volume_path, volume_size)
 
-        self._set_rw_permissions_for_all(volume_path)
+        self._set_rw_permissions(volume_path)
 
     def _ensure_shares_mounted(self):
         """Look for remote shares in the flags and tries to mount them
@@ -197,12 +243,12 @@ class RemoteFSDriver(driver.VolumeDriver):
     def _delete(self, path):
         # Note(lpetrut): this method is needed in order to provide
         # interoperability with Windows as it will be overridden.
-        self._execute('rm', '-f', path, run_as_root=True)
+        self._execute('rm', '-f', path, run_as_root=self._execute_as_root)
 
     def _create_sparsed_file(self, path, size):
         """Creates file with 0 disk usage."""
         self._execute('truncate', '-s', '%sG' % size,
-                      path, run_as_root=True)
+                      path, run_as_root=self._execute_as_root)
 
     def _create_regular_file(self, path, size):
         """Creates regular file of given size. Takes a lot of time for large
@@ -215,7 +261,7 @@ class RemoteFSDriver(driver.VolumeDriver):
         self._execute('dd', 'if=/dev/zero', 'of=%s' % path,
                       'bs=%dM' % block_size_mb,
                       'count=%d' % block_count,
-                      run_as_root=True)
+                      run_as_root=self._execute_as_root)
 
     def _create_qcow2_file(self, path, size_gb):
         """Creates a QCOW2 file of a given size."""
@@ -223,15 +269,39 @@ class RemoteFSDriver(driver.VolumeDriver):
         self._execute('qemu-img', 'create', '-f', 'qcow2',
                       '-o', 'preallocation=metadata',
                       path, str(size_gb * units.Gi),
-                      run_as_root=True)
+                      run_as_root=self._execute_as_root)
+
+    def _set_rw_permissions(self, path):
+        """Sets access permissions for given NFS path.
+
+        Volume file permissions are set based upon the value of
+        secure_file_permissions: 'true' sets secure access permissions and
+        'false' sets more open (insecure) access permissions.
+
+        :param path: the volume file path.
+        """
+        if self.configuration.nas_secure_file_permissions == 'true':
+            permissions = '660'
+            LOG.debug('File path %s is being set with permissions: %s' %
+                      (path, permissions))
+        else:
+            permissions = 'ugo+rw'
+            parms = {'path': path, 'perm': permissions}
+            LOG.warn(_('%(path)s is being set with open permissions: '
+                       '%(perm)s') % parms)
+
+        self._execute('chmod', permissions, path,
+                      run_as_root=self._execute_as_root)
 
     def _set_rw_permissions_for_all(self, path):
         """Sets 666 permissions for the path."""
-        self._execute('chmod', 'ugo+rw', path, run_as_root=True)
+        self._execute('chmod', 'ugo+rw', path,
+                      run_as_root=self._execute_as_root)
 
     def _set_rw_permissions_for_owner(self, path):
         """Sets read-write permissions to the owner for the path."""
-        self._execute('chmod', 'u+rw', path, run_as_root=True)
+        self._execute('chmod', 'u+rw', path,
+                      run_as_root=self._execute_as_root)
 
     def local_path(self, volume):
         """Get volume path (mounted locally fs path) for given volume
@@ -243,12 +313,15 @@ class RemoteFSDriver(driver.VolumeDriver):
 
     def copy_image_to_volume(self, context, volume, image_service, image_id):
         """Fetch the image from image_service and write it to the volume."""
+        run_as_root = self._execute_as_root
+
         image_utils.fetch_to_raw(context,
                                  image_service,
                                  image_id,
                                  self.local_path(volume),
                                  self.configuration.volume_dd_blocksize,
-                                 size=volume['size'])
+                                 size=volume['size'],
+                                 run_as_root=run_as_root)
 
         # NOTE (leseb): Set the virtual size of the image
         # the raw conversion overwrote the destination file
@@ -257,9 +330,11 @@ class RemoteFSDriver(driver.VolumeDriver):
         # thus the initial 'size' parameter is not honored
         # this sets the size to the one asked in the first place by the user
         # and then verify the final virtual size
-        image_utils.resize_image(self.local_path(volume), volume['size'])
+        image_utils.resize_image(self.local_path(volume), volume['size'],
+                                 run_as_root=run_as_root)
 
-        data = image_utils.qemu_img_info(self.local_path(volume))
+        data = image_utils.qemu_img_info(self.local_path(volume),
+                                         run_as_root=run_as_root)
         virt_size = data.virtual_size / units.Gi
         if virt_size != volume['size']:
             raise exception.ImageUnacceptable(
@@ -375,6 +450,88 @@ class RemoteFSDriver(driver.VolumeDriver):
     def _ensure_share_mounted(self, share):
         raise NotImplementedError()
 
+    def secure_file_operations_enabled(self):
+        """Determine if driver is operating in Secure File Operations mode.
+
+        The Cinder Volume driver needs to query if this driver is operating
+        in a secure file mode; check our nas_secure_file_operations flag.
+        """
+        if self.configuration.nas_secure_file_operations == 'true':
+            return True
+        return False
+
+    def set_nas_security_options(self, is_new_cinder_install):
+        """Determine the setting to use for Secure NAS options.
+
+        This method must be overridden by child wishing to use secure
+        NAS file operations. This base method will set the NAS security
+        options to false.
+        """
+        doc_html = "http://docs.openstack.org/admin-guide-cloud/content" \
+                   "/nfs_backend.html"
+        self.configuration.nas_secure_file_operations = 'false'
+        LOG.warn(_("The NAS file operations will be run as root: allowing "
+                   "root level access at the storage backend. This is "
+                   "considered an insecure NAS environment. Please see %s for "
+                   "information on a secure NAS configuration.") %
+                 doc_html)
+        self.configuration.nas_secure_file_permissions = 'false'
+        LOG.warn(_("The NAS file permissions mode will be 666 (allowing "
+                   "other/world read & write access). This is considered an "
+                   "insecure NAS environment. Please see %s for information "
+                   "on a secure NFS configuration.") %
+                 doc_html)
+
+    def _determine_nas_security_option_setting(self, nas_option, mount_point,
+                                               is_new_cinder_install):
+        """Determine NAS security option setting when 'auto' is assigned.
+
+        This method determines the final 'true'/'false' setting of an NAS
+        security option when the default value of 'auto' has been detected.
+        If the nas option isn't 'auto' then its current value is used.
+
+        :param nas_option: The NAS security option value loaded from config.
+        :param mount_point: Mount where indicator file is written.
+        :param is_new_cinder_install: boolean for new Cinder installation.
+        :return string: 'true' or 'false' for new option setting.
+        """
+        if nas_option == 'auto':
+            # For auto detection, we first check to see if we have been
+            # through this process before by checking for the existence of
+            # the Cinder secure environment indicator file.
+            file_name = '.cinderSecureEnvIndicator'
+            file_path = os.path.join(mount_point, file_name)
+            if os.path.isfile(file_path):
+                nas_option = 'true'
+                LOG.info(_('Cinder secure environment indicator file exists.'))
+            else:
+                # The indicator file does not exist. If it is a new
+                # installation, set to 'true' and create the indicator file.
+                if is_new_cinder_install:
+                    nas_option = 'true'
+                    try:
+                        with open(file_path, 'w') as fh:
+                            fh.write('Detector file for Cinder secure '
+                                     'environment usage.\n')
+                            fh.write('Do not delete this file.\n')
+
+                        # Set the permissions on our special marker file to
+                        # protect from accidental removal (owner write only).
+                        self._execute('chmod', '640', file_path,
+                                      run_as_root=False)
+                        LOG.info(_('New Cinder secure environment indicator '
+                                   'file created at path %s.') % file_path)
+                    except IOError as err:
+                        LOG.error(_('Failed to created Cinder secure '
+                                    'environment indicator file: %s') %
+                                  format(err))
+                else:
+                    # For existing installs, we default to 'false'. The
+                    # admin can always set the option at the driver config.
+                    nas_option = 'false'
+
+        return nas_option
+
 
 class RemoteFSSnapDriver(RemoteFSDriver):
     """Base class for remotefs drivers implementing qcow2 snapshots.
@@ -455,12 +612,13 @@ class RemoteFSSnapDriver(RemoteFSDriver):
         raise NotImplementedError()
 
     def _img_commit(self, path):
-        self._execute('qemu-img', 'commit', path, run_as_root=True)
+        self._execute('qemu-img', 'commit', path,
+                      run_as_root=self._execute_as_root)
         self._delete(path)
 
     def _rebase_img(self, image, backing_file, volume_format):
         self._execute('qemu-img', 'rebase', '-u', '-b', backing_file, image,
-                      '-F', volume_format, run_as_root=True)
+                      '-F', volume_format, run_as_root=self._execute_as_root)
 
     def _read_info_file(self, info_path, empty_if_missing=False):
         """Return dict of snapshot information.
@@ -530,7 +688,8 @@ class RemoteFSSnapDriver(RemoteFSDriver):
         mount_point = self._get_mount_point_for_share(share)
 
         out, _ = self._execute('df', '--portability', '--block-size', '1',
-                               mount_point, run_as_root=True)
+                               mount_point,
+                               run_as_root=self._execute_as_root)
         out = out.splitlines()[1]
 
         size = int(out.split()[1])
@@ -885,7 +1044,7 @@ class RemoteFSSnapDriver(RemoteFSDriver):
 
         command = ['qemu-img', 'create', '-f', 'qcow2', '-o',
                    'backing_file=%s' % backing_path_full_path, new_snap_path]
-        self._execute(*command, run_as_root=True)
+        self._execute(*command, run_as_root=self._execute_as_root)
 
         info = self._qemu_img_info(backing_path_full_path,
                                    snapshot['volume']['name'])
@@ -895,9 +1054,9 @@ class RemoteFSSnapDriver(RemoteFSDriver):
                    '-b', backing_filename,
                    '-F', backing_fmt,
                    new_snap_path]
-        self._execute(*command, run_as_root=True)
+        self._execute(*command, run_as_root=self._execute_as_root)
 
-        self._set_rw_permissions_for_all(new_snap_path)
+        self._set_rw_permissions(new_snap_path)
 
     def _create_snapshot(self, snapshot):
         """Create a snapshot.

cinder/volume/manager.py

@@ -13,6 +13,7 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
+
 """
 Volume manager manages creating, attaching, detaching, and persistent storage.
 
@@ -176,11 +177,17 @@ class VolumeManager(manager.SchedulerDependentManager):
             LOG.warn(_("Driver path %s is deprecated, update your "
                        "configuration to the new path."), volume_driver)
             volume_driver = MAPPING[volume_driver]
+
+        vol_db_empty = self._set_voldb_empty_at_startup_indicator(
+            context.get_admin_context())
+        LOG.debug("Cinder Volume DB check: vol_db_empty=%s" % vol_db_empty)
+
         self.driver = importutils.import_object(
             volume_driver,
             configuration=self.configuration,
             db=self.db,
-            host=self.host)
+            host=self.host,
+            is_vol_db_empty=vol_db_empty)
 
         self.driver = profiler.trace_cls("driver")(self.driver)
         try:
@@ -237,6 +244,24 @@ class VolumeManager(manager.SchedulerDependentManager):
         self.stats['pools'][pool]['allocated_capacity_gb'] = pool_sum
         self.stats['allocated_capacity_gb'] += volume['size']
 
+    def _set_voldb_empty_at_startup_indicator(self, ctxt):
+        """Determine if the Cinder volume DB is empty.
+
+        A check of the volume DB is done to determine whether it is empty or
+        not at this point.
+
+        :param ctxt: our working context
+        """
+        vol_entries = self.db.volume_get_all(ctxt, None, 1, 'created_at',
+                                             None, filters=None)
+
+        if len(vol_entries) == 0:
+            LOG.info(_("Determined volume DB was empty at startup."))
+            return True
+        else:
+            LOG.info(_("Determined volume DB was not empty at startup."))
+            return False
+
     def init_host(self):
         """Do any initialization that needs to be run if this is a
            standalone service.

etc/cinder/cinder.conf.sample

@@ -1466,6 +1466,23 @@
 # (string value)
 #nas_private_key=
 
+# Allow network-attached storage systems to operate in a
+# secure environment where root level access is not permitted.
+# If set to False, access is as the root user and insecure. If
+# set to True, access is not as root. If set to auto, a check
+# is done to determine if this is a new installation: True is
+# used if so, otherwise False. Default is auto. (string value)
+#nas_secure_file_operations=auto
+
+# Set more secure file permissions on network-attached storage
+# volume files to restrict broad other/world access. If set to
+# False, volumes are created with open permissions. If set to
+# True, volumes are created with permissions for the cinder
+# user and group (660). If set to auto, a check is done to
+# determine if this is a new installation: True is used if so,
+# otherwise False. Default is auto. (string value)
+#nas_secure_file_permissions=auto
+
 # IBMNAS platform type to be used as backend storage; valid
 # values are - v7ku : for using IBM Storwize V7000 Unified,
 # sonas : for using IBM Scale Out NAS, gpfs-nas : for using
@@ -1904,6 +1921,23 @@
 # (string value)
 #nas_private_key=
 
+# Allow network-attached storage systems to operate in a
+# secure environment where root level access is not permitted.
+# If set to False, access is as the root user and insecure. If
+# set to True, access is not as root. If set to auto, a check
+# is done to determine if this is a new installation: True is
+# used if so, otherwise False. Default is auto. (string value)
+#nas_secure_file_operations=auto
+
+# Set more secure file permissions on network-attached storage
+# volume files to restrict broad other/world access. If set to
+# False, volumes are created with open permissions. If set to
+# True, volumes are created with permissions for the cinder
+# user and group (660). If set to auto, a check is done to
+# determine if this is a new installation: True is used if so,
+# otherwise False. Default is auto. (string value)
+#nas_secure_file_permissions=auto
+
 
 #
 # Options defined in cinder.volume.drivers.san.hp.hp_3par_common