CVE-2013-4183

@@ -1439,7 +1439,7 @@ def test_delete_busy_volume(self):
         self.stubs.Set(self.volume.driver, '_volume_not_present',
                        lambda x: False)
         self.stubs.Set(self.volume.driver, '_delete_volume',
-                       lambda x, y: False)
+                       lambda x: False)
         # Want DriverTestCase._fake_execute to return 'o' so that
         # volume.driver.delete_volume() raises the VolumeIsBusy exception.
         self.output = 'o'
@@ -1490,6 +1490,29 @@ def test_convert_blocksize_option(self):
         self.assertEquals(bs, '1M')
         self.assertEquals(count, 1024)
 
+    def test_clear_volume(self):
+        configuration = conf.Configuration(fake_opt, 'fake_group')
+        configuration.volume_clear = 'zero'
+        configuration.volume_clear_size = 0
+        lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
+        self.stubs.Set(lvm_driver, '_copy_volume', lambda *a, **kw: True)
+
+        fake_volume = {'name': 'test1',
+                       'volume_name': 'test1',
+                       'id': 'test1'}
+
+        # Test volume has 'size' field
+        volume = dict(fake_volume, size='123')
+        self.assertEquals(True, lvm_driver.clear_volume(volume))
+
+        # Test volume has 'volume_size' field
+        volume = dict(fake_volume, volume_size='123')
+        self.assertEquals(True, lvm_driver.clear_volume(volume))
+
+        # Test volume without 'size' field and 'volume_size' field
+        volume = dict(fake_volume)
+        self.assertEquals(None, lvm_driver.clear_volume(volume))
+
 
 class ISCSITestCase(DriverTestCase):
     """Test Case for ISCSIDriver"""

@@ -162,7 +162,7 @@ def _volume_not_present(self, volume_name):
             return True
         return False
 
-    def _delete_volume(self, volume, size_in_g):
+    def _delete_volume(self, volume):
         """Deletes a logical volume."""
         # zero out old volumes to prevent data leaking between users
         # TODO(ja): reclaiming space should be done lazy and low priority
@@ -218,19 +218,18 @@ def delete_volume(self, volume):
             if (out[0] == 'o') or (out[0] == 'O'):
                 raise exception.VolumeIsBusy(volume_name=volume['name'])
 
-        self._delete_volume(volume, volume['size'])
+        self._delete_volume(volume)
 
     def clear_volume(self, volume):
         """unprovision old volumes to prevent data leaking between users."""
 
         vol_path = self.local_path(volume)
-        size_in_g = volume.get('size')
-        size_in_m = self.configuration.volume_clear_size
-
-        if not size_in_g:
+        size_in_g = volume.get('size', volume.get('volume_size', None))
+        if size_in_g is None:
             LOG.warning(_("Size for volume: %s not found, "
-                          "skipping secure delete.") % volume['name'])
+                          "skipping secure delete.") % volume['id'])
             return
+        size_in_m = self.configuration.volume_clear_size
 
         if self.configuration.volume_clear == 'none':
             return
@@ -275,7 +274,7 @@ def delete_snapshot(self, snapshot):
 
         # TODO(yamahata): zeroing out the whole snapshot triggers COW.
         # it's quite slow.
-        self._delete_volume(snapshot, snapshot['volume_size'])
+        self._delete_volume(snapshot)
 
     def local_path(self, volume):
         # NOTE(vish): stops deprecation warning

@@ -24,6 +24,7 @@
 import os
 
 import mox
+from oslo.config import cfg
 import shutil
 import tempfile
 
@@ -43,11 +44,16 @@
 from cinder.tests.image import fake as fake_image
 from cinder.volume import configuration as conf
 from cinder.volume import driver
+from cinder.volume.drivers import lvm
 from cinder.volume import iscsi
 
 QUOTAS = quota.QUOTAS
 FLAGS = flags.FLAGS
 
+fake_opt = [
+    cfg.StrOpt('fake_opt', default='fake', help='fake opts')
+]
+
 
 class VolumeTestCase(test.TestCase):
     """Test Case for volumes."""
@@ -907,7 +913,7 @@ def test_delete_busy_volume(self):
         self.stubs.Set(self.volume.driver, '_volume_not_present',
                        lambda x: False)
         self.stubs.Set(self.volume.driver, '_delete_volume',
-                       lambda x, y: False)
+                       lambda x: False)
         # Want DriverTestCase._fake_execute to return 'o' so that
         # volume.driver.delete_volume() raises the VolumeIsBusy exception.
         self.output = 'o'
@@ -920,6 +926,34 @@ def test_delete_busy_volume(self):
         self.volume.driver.delete_volume({'name': 'test1', 'size': 1024})
 
 
+class LVMVolumeDriverTestCase(DriverTestCase):
+    """Test case for VolumeDriver"""
+    driver_name = "cinder.volume.drivers.lvm.LVMVolumeDriver"
+
+    def test_clear_volume(self):
+        configuration = conf.Configuration(fake_opt, 'fake_group')
+        configuration.volume_clear = 'zero'
+        configuration.volume_clear_size = 0
+        lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
+        self.stubs.Set(lvm_driver, '_copy_volume', lambda *a, **kw: True)
+
+        fake_volume = {'name': 'test1',
+                       'volume_name': 'test1',
+                       'id': 'test1'}
+
+        # Test volume has 'size' field
+        volume = dict(fake_volume, size='123')
+        self.assertEquals(True, lvm_driver.clear_volume(volume))
+
+        # Test volume has 'volume_size' field
+        volume = dict(fake_volume, volume_size='123')
+        self.assertEquals(True, lvm_driver.clear_volume(volume))
+
+        # Test volume without 'size' field and 'volume_size' field
+        volume = dict(fake_volume)
+        self.assertEquals(None, lvm_driver.clear_volume(volume))
+
+
 class ISCSITestCase(DriverTestCase):
     """Test Case for ISCSIDriver"""
     driver_name = "cinder.volume.drivers.lvm.LVMISCSIDriver"

@@ -125,7 +125,7 @@ def _volume_not_present(self, volume_name):
             return True
         return False
 
-    def _delete_volume(self, volume, size_in_g):
+    def _delete_volume(self, volume):
         """Deletes a logical volume."""
         # zero out old volumes to prevent data leaking between users
         # TODO(ja): reclaiming space should be done lazy and low priority
@@ -180,19 +180,18 @@ def delete_volume(self, volume):
             if (out[0] == 'o') or (out[0] == 'O'):
                 raise exception.VolumeIsBusy(volume_name=volume['name'])
 
-        self._delete_volume(volume, volume['size'])
+        self._delete_volume(volume)
 
     def clear_volume(self, volume):
         """unprovision old volumes to prevent data leaking between users."""
 
         vol_path = self.local_path(volume)
-        size_in_g = volume.get('size')
-        size_in_m = self.configuration.volume_clear_size
-
-        if not size_in_g:
+        size_in_g = volume.get('size', volume.get('volume_size', None))
+        if size_in_g is None:
             LOG.warning(_("Size for volume: %s not found, "
-                          "skipping secure delete.") % volume['name'])
+                          "skipping secure delete.") % volume['id'])
             return
+        size_in_m = self.configuration.volume_clear_size
 
         if self.configuration.volume_clear == 'none':
             return
@@ -237,7 +236,7 @@ def delete_snapshot(self, snapshot):
 
         # TODO(yamahata): zeroing out the whole snapshot triggers COW.
         # it's quite slow.
-        self._delete_volume(snapshot, snapshot['volume_size'])
+        self._delete_volume(snapshot)
 
     def local_path(self, volume):
         # NOTE(vish): stops deprecation warning