openstack/openstacksdk
Code Issues Proposed changes

Merge "encapsulate md5 calls for fips"

Browse Source
This commit is contained in:
Zuul
2021-02-15 22:21:00 +00:00
committed by Gerrit Code Review
parent f8a323ff1f 2753692813
commit 58b28ee592
6 changed files with 112 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
openstack/cloud/_object_store.py
View File

@@ -28,6 +28,7 @@ from openstack.cloud import _normalize
from openstack.cloud import _utils
from openstack import exceptions
from openstack import proxy
from openstack import utils
DEFAULT_OBJECT_SEGMENT_SIZE = 1073741824 # 1GB
@@ -218,7 +219,7 @@ class ObjectStoreCloudMixin(_normalize.Normalizer):
self._file_hash_cache[file_key]['sha256'])
def _calculate_data_hashes(self, data):
md5 = hashlib.md5()
md5 = utils.md5(usedforsecurity=False)
sha256 = hashlib.sha256()
if hasattr(data, 'read'):

6
openstack/image/_download.py
View File

@@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import io
import hashlib
from openstack import exceptions
from openstack import utils
@@ -45,7 +44,7 @@ class DownloadMixin:
details = self.fetch(session)
checksum = details.checksum
md5 = hashlib.md5()
md5 = utils.md5(usedforsecurity=False)
if output:
try:
if isinstance(output, io.IOBase):
@@ -73,7 +72,8 @@ class DownloadMixin:
return resp
if checksum is not None:
_verify_checksum(hashlib.md5(resp.content), checksum)
_verify_checksum(utils.md5(resp.content, usedforsecurity=False),
checksum)
else:
session.log.warning(
"Unable to verify the integrity of image %s", (self.id))

3
openstack/tests/fakes.py
View File

@@ -24,6 +24,7 @@ import uuid
from openstack.orchestration.util import template_format
from openstack.cloud import meta
from openstack import utils
PROJECT_ID = '1c36b64c840a42cd9e9b931a369337f0'
FLAVOR_ID = u'0c1d9008-f546-4608-9e8f-f8bdaec8dddd'
@@ -225,7 +226,7 @@ def make_fake_image(
data=None,
checksum=u'ee36e35a297980dee1b514de9803ec6d'):
if data:
md5 = hashlib.md5()
md5 = utils.md5(usedforsecurity=False)
sha256 = hashlib.sha256()
with open(data, 'rb') as file_obj:
for chunk in iter(lambda: file_obj.read(8192), b''):

4
openstack/tests/unit/image/v2/test_image.py
View File

@@ -9,7 +9,6 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import hashlib
import io
import operator
import tempfile
@@ -22,6 +21,7 @@ from openstack import _log
from openstack import exceptions
from openstack.image.v2 import image
from openstack.tests.unit import base
from openstack import utils
IDENTIFIER = 'IDENTIFIER'
EXAMPLE = {
@@ -89,7 +89,7 @@ EXAMPLE = {
def calculate_md5_checksum(data):
checksum = hashlib.md5()
checksum = utils.md5(usedforsecurity=False)
for chunk in data:
checksum.update(chunk)
return checksum.hexdigest()

86
openstack/tests/unit/test_utils.py
View File

@@ -13,8 +13,10 @@
# under the License.
import concurrent.futures
import hashlib
import logging
from unittest import mock
from unittest import skipIf
import sys
import fixtures
@@ -302,3 +304,87 @@ class TestTinyDAG(base.TestCase):
def test_walker_fn(graph, node, lst):
lst.append(node)
graph.node_done(node)
class Test_md5(base.TestCase):
def setUp(self):
super(Test_md5, self).setUp()
self.md5_test_data = "Openstack forever".encode('utf-8')
try:
self.md5_digest = hashlib.md5( # nosec
self.md5_test_data).hexdigest()
self.fips_enabled = False
except ValueError:
self.md5_digest = '0d6dc3c588ae71a04ce9a6beebbbba06'
self.fips_enabled = True
def test_md5_with_data(self):
if not self.fips_enabled:
digest = utils.md5(self.md5_test_data).hexdigest()
self.assertEqual(digest, self.md5_digest)
else:
# on a FIPS enabled system, this throws a ValueError:
# [digital envelope routines: EVP_DigestInit_ex] disabled for FIPS
self.assertRaises(ValueError, utils.md5, self.md5_test_data)
if not self.fips_enabled:
digest = utils.md5(self.md5_test_data,
usedforsecurity=True).hexdigest()
self.assertEqual(digest, self.md5_digest)
else:
self.assertRaises(
ValueError, utils.md5, self.md5_test_data,
usedforsecurity=True)
digest = utils.md5(self.md5_test_data,
usedforsecurity=False).hexdigest()
self.assertEqual(digest, self.md5_digest)
def test_md5_without_data(self):
if not self.fips_enabled:
test_md5 = utils.md5()
test_md5.update(self.md5_test_data)
digest = test_md5.hexdigest()
self.assertEqual(digest, self.md5_digest)
else:
self.assertRaises(ValueError, utils.md5)
if not self.fips_enabled:
test_md5 = utils.md5(usedforsecurity=True)
test_md5.update(self.md5_test_data)
digest = test_md5.hexdigest()
self.assertEqual(digest, self.md5_digest)
else:
self.assertRaises(ValueError, utils.md5, usedforsecurity=True)
test_md5 = utils.md5(usedforsecurity=False)
test_md5.update(self.md5_test_data)
digest = test_md5.hexdigest()
self.assertEqual(digest, self.md5_digest)
@skipIf(sys.version_info.major == 2,
"hashlib.md5 does not raise TypeError here in py2")
def test_string_data_raises_type_error(self):
if not self.fips_enabled:
self.assertRaises(TypeError, hashlib.md5, u'foo')
self.assertRaises(TypeError, utils.md5, u'foo')
self.assertRaises(
TypeError, utils.md5, u'foo', usedforsecurity=True)
else:
self.assertRaises(ValueError, hashlib.md5, u'foo')
self.assertRaises(ValueError, utils.md5, u'foo')
self.assertRaises(
ValueError, utils.md5, u'foo', usedforsecurity=True)
self.assertRaises(
TypeError, utils.md5, u'foo', usedforsecurity=False)
def test_none_data_raises_type_error(self):
if not self.fips_enabled:
self.assertRaises(TypeError, hashlib.md5, None)
self.assertRaises(TypeError, utils.md5, None)
self.assertRaises(
TypeError, utils.md5, None, usedforsecurity=True)
else:
self.assertRaises(ValueError, hashlib.md5, None)
self.assertRaises(ValueError, utils.md5, None)
self.assertRaises(
ValueError, utils.md5, None, usedforsecurity=True)
self.assertRaises(
TypeError, utils.md5, None, usedforsecurity=False)

17
openstack/utils.py
View File

@@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import hashlib
import queue
import string
import threading
@@ -232,6 +233,22 @@ def maximum_supported_microversion(adapter, client_maximum):
return discover.version_to_string(result)
try:
_test_md5 = hashlib.md5(usedforsecurity=False) # nosec
# Python distributions that support a hashlib.md5 with the usedforsecurity
# keyword can just use that md5 definition as-is
# See https://bugs.python.org/issue9216
md5 = hashlib.md5
except TypeError:
def md5(string=b'', usedforsecurity=True):
"""Return an md5 hashlib object without usedforsecurity parameter
For python distributions that do not yet support this keyword
parameter, we drop the parameter
"""
return hashlib.md5(string) # nosec
class TinyDAG:
"""Tiny DAG