openstack/swift
Code Issues Proposed changes

formpost: deprecate sha1 signatures

Browse Source 
We've known this would eventually be necessary for a while [1], and
way back in 2017 we started seeing SHA-1 collisions [2].

This patch follows the approach of soft deprecation of SHA1 in tempurl.
It's still a default digest, but we'll start with warning as the
middleware is loaded and exposing any deprecated digests
(if they're still allowed) in /info.

Further, because there is much shared code between formpost and tempurl, this
patch also goes and refactors shared code out into swift.common.digest.
Now that we have a digest, we also move digest related code:
 - get_hmac
 - extract_digest_and_algorithm

[1] https://www.schneier.com/blog/archives/2012/10/when_will_we_se.html
[2] https://security.googleblog.com/2017/02/announcing-first-sha1-collision.html

Change-Id: I581cadd6bc79e623f1dae071025e4d375254c1d9
This commit is contained in:
Matthew Oliver 2022-04-19 15:26:11 +10:00
parent 25b6bd9f2c
commit 2d063cd61f
11 changed files with 431 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
doc/source/misc.rst
View File

@ -39,6 +39,17 @@ Container Sync Realms
:members:
:show-inheritance:
.. _digest:
Digest
======
.. automodule:: swift.common.digest
:members:
:undoc-members:
:show-inheritance:
.. _direct_client:
Direct Client

151
swift/common/digest.py Normal file
View File

@ -0,0 +1,151 @@
# Copyright (c) 2022 NVIDIA
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT 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 binascii
import hashlib
import hmac
import six
from swift.common.utils import strict_b64decode
DEFAULT_ALLOWED_DIGESTS = 'sha1 sha256 sha512'
DEPRECATED_DIGESTS = {'sha1'}
SUPPORTED_DIGESTS = set(DEFAULT_ALLOWED_DIGESTS.split()) | DEPRECATED_DIGESTS
def get_hmac(request_method, path, expires, key, digest="sha1",
ip_range=None):
"""
Returns the hexdigest string of the HMAC (see RFC 2104) for
the request.
:param request_method: Request method to allow.
:param path: The path to the resource to allow access to.
:param expires: Unix timestamp as an int for when the URL
expires.
:param key: HMAC shared secret.
:param digest: constructor or the string name for the digest to use in
calculating the HMAC
Defaults to SHA1
:param ip_range: The ip range from which the resource is allowed
to be accessed. We need to put the ip_range as the
first argument to hmac to avoid manipulation of the path
due to newlines being valid in paths
e.g. /v1/a/c/o\\n127.0.0.1
:returns: hexdigest str of the HMAC for the request using the specified
digest algorithm.
"""
# These are the three mandatory fields.
parts = [request_method, str(expires), path]
formats = [b"%s", b"%s", b"%s"]
if ip_range:
parts.insert(0, ip_range)
formats.insert(0, b"ip=%s")
if not isinstance(key, six.binary_type):
key = key.encode('utf8')
message = b'\n'.join(
fmt % (part if isinstance(part, six.binary_type)
else part.encode("utf-8"))
for fmt, part in zip(formats, parts))
if six.PY2 and isinstance(digest, six.string_types):
digest = getattr(hashlib, digest)
return hmac.new(key, message, digest).hexdigest()
def get_allowed_digests(conf_digests, logger=None):
"""
Pulls out 'allowed_digests' from the supplied conf. Then compares them with
the list of supported and deprecated digests and returns whatever remain.
When something is unsupported or deprecated it'll log a warning.
:param conf_digests: iterable of allowed digests. If empty, defaults to
DEFAULT_ALLOWED_DIGESTS.
:param logger: optional logger; if provided, use it issue deprecation
warnings
:returns: A set of allowed digests that are supported and a set of
deprecated digests.
:raises: ValueError, if there are no digests left to return.
"""
allowed_digests = set(digest.lower() for digest in conf_digests)
if not allowed_digests:
allowed_digests = SUPPORTED_DIGESTS
not_supported = allowed_digests - SUPPORTED_DIGESTS
if not_supported:
if logger:
logger.warning('The following digest algorithms are configured '
'but not supported: %s', ', '.join(not_supported))
allowed_digests -= not_supported
deprecated = allowed_digests & DEPRECATED_DIGESTS
if deprecated and logger:
if not conf_digests:
logger.warning('The following digest algorithms are allowed by '
'default but deprecated: %s. Support will be '
'disabled by default in a future release, and '
'later removed entirely.', ', '.join(deprecated))
else:
logger.warning('The following digest algorithms are configured '
'but deprecated: %s. Support will be removed in a '
'future release.', ', '.join(deprecated))
if not allowed_digests:
raise ValueError('No valid digest algorithms are configured')
return allowed_digests, deprecated
def extract_digest_and_algorithm(value):
"""
Returns a tuple of (digest_algorithm, hex_encoded_digest)
from a client-provided string of the form::
<hex-encoded digest>
or::
<algorithm>:<base64-encoded digest>
Note that hex-encoded strings must use one of sha1, sha256, or sha512.
:raises: ValueError on parse failures
"""
if ':' in value:
algo, value = value.split(':', 1)
# accept both standard and url-safe base64
if ('-' in value or '_' in value) and not (
'+' in value or '/' in value):
value = value.replace('-', '+').replace('_', '/')
value = binascii.hexlify(strict_b64decode(value + '=='))
if not six.PY2:
value = value.decode('ascii')
else:
try:
binascii.unhexlify(value) # make sure it decodes
except TypeError:
# This is just for py2
raise ValueError('Non-hexadecimal digit found')
algo = {
40: 'sha1',
64: 'sha256',
128: 'sha512',
}.get(len(value))
if not algo:
raise ValueError('Bad digest length')
return algo, value

23
swift/common/middleware/formpost.py
View File

@ -131,11 +131,12 @@ from six.moves.urllib.parse import quote
from swift.common.constraints import valid_api_version
from swift.common.exceptions import MimeInvalid
from swift.common.middleware.tempurl import get_tempurl_keys_from_metadata, \
SUPPORTED_DIGESTS
from swift.common.middleware.tempurl import get_tempurl_keys_from_metadata
from swift.common.digest import get_allowed_digests, \
extract_digest_and_algorithm, DEFAULT_ALLOWED_DIGESTS
from swift.common.utils import streq_const_time, parse_content_disposition, \
parse_mime_headers, iter_multipart_mime_documents, reiterate, \
close_if_possible, get_logger, extract_digest_and_algorithm
close_if_possible, get_logger
from swift.common.registry import register_swift_info
from swift.common.wsgi import make_pre_authed_env
from swift.common.swob import HTTPUnauthorized, wsgi_to_str, str_to_wsgi
@ -216,7 +217,7 @@ class FormPost(object):
# deprecate sha1 yet. We'll change this to DEFAULT_ALLOWED_DIGESTS
# later.
self.allowed_digests = conf.get(
'allowed_digests', SUPPORTED_DIGESTS)
'allowed_digests', DEFAULT_ALLOWED_DIGESTS.split())
def __call__(self, env, start_response):
"""
@ -484,17 +485,11 @@ def filter_factory(global_conf, **local_conf):
conf.update(local_conf)
logger = get_logger(conf, log_route='formpost')
allowed_digests = set(conf.get('allowed_digests', '').split()) or \
SUPPORTED_DIGESTS
not_supported = allowed_digests - SUPPORTED_DIGESTS
if not_supported:
logger.warning('The following digest algorithms are configured but '
'not supported: %s', ', '.join(not_supported))
allowed_digests -= not_supported
if not allowed_digests:
raise ValueError('No valid digest algorithms are configured '
'for formpost')
allowed_digests, deprecated_digests = get_allowed_digests(
conf.get('allowed_digests', '').split(), logger)
info = {'allowed_digests': sorted(allowed_digests)}
if deprecated_digests:
info['deprecated_digests'] = sorted(deprecated_digests)
register_swift_info('formpost', **info)
conf.update(info)
return lambda app: FormPost(app, conf)

36
swift/common/middleware/tempurl.py
View File

@ -309,10 +309,12 @@ from six.moves.urllib.parse import urlencode
from swift.proxy.controllers.base import get_account_info, get_container_info
from swift.common.header_key_dict import HeaderKeyDict
from swift.common.digest import get_allowed_digests, \
extract_digest_and_algorithm, DEFAULT_ALLOWED_DIGESTS, get_hmac
from swift.common.swob import header_to_environ_key, HTTPUnauthorized, \
HTTPBadRequest, wsgi_to_str
from swift.common.utils import split_path, get_valid_utf8_str, \
get_hmac, streq_const_time, quote, get_logger, extract_digest_and_algorithm
streq_const_time, quote, get_logger
from swift.common.registry import register_swift_info, register_sensitive_param
@ -340,10 +342,6 @@ DEFAULT_OUTGOING_REMOVE_HEADERS = 'x-object-meta-*'
#: '*' to indicate a prefix match.
DEFAULT_OUTGOING_ALLOW_HEADERS = 'x-object-meta-public-*'
DEFAULT_ALLOWED_DIGESTS = 'sha1 sha256 sha512'
DEPRECATED_DIGESTS = {'sha1'}
SUPPORTED_DIGESTS = set(DEFAULT_ALLOWED_DIGESTS.split()) | DEPRECATED_DIGESTS
CONTAINER_SCOPE = 'container'
ACCOUNT_SCOPE = 'account'
@ -841,34 +839,14 @@ def filter_factory(global_conf, **local_conf):
'incoming_allow_headers': DEFAULT_INCOMING_ALLOW_HEADERS,
'outgoing_remove_headers': DEFAULT_OUTGOING_REMOVE_HEADERS,
'outgoing_allow_headers': DEFAULT_OUTGOING_ALLOW_HEADERS,
'allowed_digests': DEFAULT_ALLOWED_DIGESTS,
}
info_conf = {k: conf.get(k, v).split() for k, v in defaults.items()}
allowed_digests = set(digest.lower()
for digest in info_conf['allowed_digests'])
not_supported = allowed_digests - SUPPORTED_DIGESTS
if not_supported:
logger.warning('The following digest algorithms are configured but '
'not supported: %s', ', '.join(not_supported))
allowed_digests -= not_supported
deprecated = allowed_digests & DEPRECATED_DIGESTS
if deprecated:
if not conf.get('allowed_digests'):
logger.warning('The following digest algorithms are allowed by '
'default but deprecated: %s. Support will be '
'disabled by default in a future release, and '
'later removed entirely.', ', '.join(deprecated))
else:
logger.warning('The following digest algorithms are configured '
'but deprecated: %s. Support will be removed in a '
'future release.', ', '.join(deprecated))
if not allowed_digests:
raise ValueError('No valid digest algorithms are configured '
'for tempurls')
allowed_digests, deprecated_digests = get_allowed_digests(
conf.get('allowed_digests', '').split(), logger)
info_conf['allowed_digests'] = sorted(allowed_digests)
if deprecated_digests:
info_conf['deprecated_digests'] = sorted(deprecated_digests)
register_swift_info('tempurl', **info_conf)
conf.update(info_conf)

85
swift/common/utils.py
View File

@ -25,7 +25,6 @@ import errno
import fcntl
import grp
import hashlib
import hmac
import json
import math
import operator
@ -282,90 +281,6 @@ except (InvalidHashPathConfigError, IOError):
pass
def extract_digest_and_algorithm(value):
"""
Returns a tuple of (digest_algorithm, hex_encoded_digest)
from a client-provided string of the form::
<hex-encoded digest>
or::
<algorithm>:<base64-encoded digest>
Note that hex-encoded strings must use one of sha1, sha256, or sha512.
:raises: ValueError on parse failures
"""
if ':' in value:
algo, value = value.split(':', 1)
# accept both standard and url-safe base64
if ('-' in value or '_' in value) and not (
'+' in value or '/' in value):
value = value.replace('-', '+').replace('_', '/')
value = binascii.hexlify(strict_b64decode(value + '=='))
if not six.PY2:
value = value.decode('ascii')
else:
try:
binascii.unhexlify(value) # make sure it decodes
except TypeError:
# This is just for py2
raise ValueError('Non-hexadecimal digit found')
algo = {
40: 'sha1',
64: 'sha256',
128: 'sha512',
}.get(len(value))
if not algo:
raise ValueError('Bad digest length')
return algo, value
def get_hmac(request_method, path, expires, key, digest="sha1",
ip_range=None):
"""
Returns the hexdigest string of the HMAC (see RFC 2104) for
the request.
:param request_method: Request method to allow.
:param path: The path to the resource to allow access to.
:param expires: Unix timestamp as an int for when the URL
expires.
:param key: HMAC shared secret.
:param digest: constructor or the string name for the digest to use in
calculating the HMAC
Defaults to SHA1
:param ip_range: The ip range from which the resource is allowed
to be accessed. We need to put the ip_range as the
first argument to hmac to avoid manipulation of the path
due to newlines being valid in paths
e.g. /v1/a/c/o\\n127.0.0.1
:returns: hexdigest str of the HMAC for the request using the specified
digest algorithm.
"""
# These are the three mandatory fields.
parts = [request_method, str(expires), path]
formats = [b"%s", b"%s", b"%s"]
if ip_range:
parts.insert(0, ip_range)
formats.insert(0, b"ip=%s")
if not isinstance(key, six.binary_type):
key = key.encode('utf8')
message = b'\n'.join(
fmt % (part if isinstance(part, six.binary_type)
else part.encode("utf-8"))
for fmt, part in zip(formats, parts))
if six.PY2 and isinstance(digest, six.string_types):
digest = getattr(hashlib, digest)
return hmac.new(key, message, digest).hexdigest()
def backward(f, blocksize=4096):
"""
A generator returning lines from a file starting with the last line,

3
swift/proxy/controllers/info.py
View File

@ -16,7 +16,8 @@
import json
from time import time
from swift.common.utils import public, get_hmac, streq_const_time
from swift.common.utils import public, streq_const_time
from swift.common.digest import get_hmac
from swift.common.registry import get_swift_info
from swift.proxy.controllers.base import Controller, delay_denial
from swift.common.swob import HTTPOk, HTTPForbidden, HTTPUnauthorized

33
test/unit/common/middleware/test_formpost.py
View File

@ -28,8 +28,9 @@ from io import BytesIO
from swift.common.swob import Request, Response, wsgi_quote
from swift.common.middleware import tempauth, formpost
from swift.common.middleware.tempurl import DEFAULT_ALLOWED_DIGESTS
from swift.common.utils import split_path
from swift.common import registry
from swift.common import registry, digest as digest_utils
from swift.proxy.controllers.base import get_cache_key
from test.debug_logger import debug_logger
@ -1656,8 +1657,11 @@ class TestFormPost(unittest.TestCase):
self.assertIsNone(exc_info)
self.assertTrue(b'FormPost: invalid starting boundary' in body)
def test_redirect_allowed_and_unsupported_digests(self):
def test_redirect_allowed_deprecated_and_unsupported_digests(self):
logger = debug_logger()
def do_test(digest):
logger.clear()
key = b'abc'
sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', 'http://redirect', 1024, 10,
@ -1670,7 +1674,11 @@ class TestFormPost(unittest.TestCase):
self.app = FakeApp(iter([('201 Created', {}, b''),
('201 Created', {}, b'')]))
self.auth = tempauth.filter_factory({})(self.app)
self.formpost = formpost.filter_factory({})(self.auth)
with mock.patch('swift.common.middleware.formpost.get_logger',
return_value=logger):
self.formpost = formpost.filter_factory(
{
'allowed_digests': DEFAULT_ALLOWED_DIGESTS})(self.auth)
status = [None]
headers = [None]
exc_info = [None]
@ -1696,6 +1704,12 @@ class TestFormPost(unittest.TestCase):
self.assertEqual(len(self.app.requests), 2)
self.assertEqual(self.app.requests[0].body, b'Test File\nOne\n')
self.assertEqual(self.app.requests[1].body, b'Test\nFile\nTwo\n')
if algorithm in digest_utils.DEPRECATED_DIGESTS:
self.assertIn(
'The following digest algorithms are configured but '
'deprecated: %s. Support will be removed in a '
'future release.' % algorithm,
logger.get_lines_for_level('warning'))
# unsupported
_body, status, _headers, _exc_info = do_test("md5")
@ -2252,7 +2266,9 @@ class TestSwiftInfo(unittest.TestCase):
self.assertIn('formpost', swift_info)
info = swift_info['formpost']
self.assertIn('allowed_digests', info)
self.assertIn('deprecated_digests', info)
self.assertEqual(info['allowed_digests'], ['sha1', 'sha256', 'sha512'])
self.assertEqual(info['deprecated_digests'], ['sha1'])
def test_non_default_methods(self):
logger = debug_logger()
@ -2265,7 +2281,9 @@ class TestSwiftInfo(unittest.TestCase):
self.assertIn('formpost', swift_info)
info = swift_info['formpost']
self.assertIn('allowed_digests', info)
self.assertIn('deprecated_digests', info)
self.assertEqual(info['allowed_digests'], ['sha1', 'sha512'])
self.assertEqual(info['deprecated_digests'], ['sha1'])
warning_lines = logger.get_lines_for_level('warning')
self.assertIn(
'The following digest algorithms are configured '
@ -2274,6 +2292,15 @@ class TestSwiftInfo(unittest.TestCase):
self.assertIn('not-a-valid-digest', warning_lines[0])
self.assertIn('md5', warning_lines[0])
def test_no_deprecated_digests(self):
formpost.filter_factory({'allowed_digests': 'sha256 sha512'})
swift_info = registry.get_swift_info()
self.assertIn('formpost', swift_info)
info = swift_info['formpost']
self.assertIn('allowed_digests', info)
self.assertNotIn('deprecated_digests', info)
self.assertEqual(info['allowed_digests'], ['sha256', 'sha512'])
def test_bad_config(self):
with self.assertRaises(ValueError):
formpost.filter_factory({

19
test/unit/common/middleware/test_tempurl.py
View File

@ -1625,6 +1625,7 @@ class TestSwiftInfo(unittest.TestCase):
self.assertEqual(set(info['outgoing_allow_headers']),
set(('x-object-meta-public-*',)))
self.assertEqual(info['allowed_digests'], ['sha1', 'sha256', 'sha512'])
self.assertEqual(info['deprecated_digests'], ['sha1'])
def test_non_default_methods(self):
tempurl.filter_factory({
@ -1647,6 +1648,24 @@ class TestSwiftInfo(unittest.TestCase):
self.assertEqual(set(info['outgoing_allow_headers']),
set(('x-object-meta-*', 'content-type')))
self.assertEqual(info['allowed_digests'], ['sha1', 'sha512'])
self.assertEqual(info['deprecated_digests'], ['sha1'])
def test_no_deprecated_digests(self):
tempurl.filter_factory({'allowed_digests': 'sha256 sha512'})
swift_info = registry.get_swift_info()
self.assertIn('tempurl', swift_info)
info = swift_info['tempurl']
self.assertEqual(set(info['methods']),
set(('GET', 'HEAD', 'PUT', 'POST', 'DELETE')))
self.assertEqual(set(info['incoming_remove_headers']),
set(('x-timestamp',)))
self.assertEqual(set(info['incoming_allow_headers']), set())
self.assertEqual(set(info['outgoing_remove_headers']),
set(('x-object-meta-*',)))
self.assertEqual(set(info['outgoing_allow_headers']),
set(('x-object-meta-public-*',)))
self.assertEqual(info['allowed_digests'], ['sha256', 'sha512'])
self.assertNotIn('deprecated_digests', info)
def test_bad_config(self):
with self.assertRaises(ValueError):

191
test/unit/common/test_digest.py Normal file
View File

@ -0,0 +1,191 @@
# Copyright (c) 2022 NVIDIA
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT 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 unittest
from swift.common import digest
from test.debug_logger import debug_logger
class TestDigestUtils(unittest.TestCase):
"""Tests for swift.common.middleware.digest """
def setUp(self):
self.logger = debug_logger('test_digest_utils')
def test_get_hmac(self):
self.assertEqual(
digest.get_hmac('GET', '/path', 1, 'abc'),
'b17f6ff8da0e251737aa9e3ee69a881e3e092e2f')
def test_get_hmac_ip_range(self):
self.assertEqual(
digest.get_hmac('GET', '/path', 1, 'abc', ip_range='127.0.0.1'),
'b30dde4d2b8562b8496466c3b46b2b9ac5054461')
def test_get_hmac_ip_range_non_binary_type(self):
self.assertEqual(
digest.get_hmac(
u'GET', u'/path', 1, u'abc', ip_range=u'127.0.0.1'),
'b30dde4d2b8562b8496466c3b46b2b9ac5054461')
def test_get_hmac_digest(self):
self.assertEqual(
digest.get_hmac(u'GET', u'/path', 1, u'abc', digest='sha256'),
'64c5558394f86b042ce1e929b34907abd9d0a57f3e20cd3f93cffd83de0206a7')
self.assertEqual(
digest.get_hmac(
u'GET', u'/path', 1, u'abc', digest=hashlib.sha256),
'64c5558394f86b042ce1e929b34907abd9d0a57f3e20cd3f93cffd83de0206a7')
self.assertEqual(
digest.get_hmac(u'GET', u'/path', 1, u'abc', digest='sha512'),
'7e95af818aec1b69b53fc2cb6d69456ec64ebda6c17b8fc8b7303b78acc8ca'
'14fc4aed96c1614a8e9d6ff45a6237711d8be294cda679624825d79aa6959b'
'5229')
self.assertEqual(
digest.get_hmac(
u'GET', u'/path', 1, u'abc', digest=hashlib.sha512),
'7e95af818aec1b69b53fc2cb6d69456ec64ebda6c17b8fc8b7303b78acc8ca'
'14fc4aed96c1614a8e9d6ff45a6237711d8be294cda679624825d79aa6959b'
'5229')
def test_extract_digest_and_algorithm(self):
self.assertEqual(
digest.extract_digest_and_algorithm(
'b17f6ff8da0e251737aa9e3ee69a881e3e092e2f'),
('sha1', 'b17f6ff8da0e251737aa9e3ee69a881e3e092e2f'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'sha1:sw3eTSuFYrhJZGbDtGsrmsUFRGE='),
('sha1', 'b30dde4d2b8562b8496466c3b46b2b9ac5054461'))
# also good with '=' stripped
self.assertEqual(
digest.extract_digest_and_algorithm(
'sha1:sw3eTSuFYrhJZGbDtGsrmsUFRGE'),
('sha1', 'b30dde4d2b8562b8496466c3b46b2b9ac5054461'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'),
('sha256', 'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'sha256:uWNxIxPNQjZpb7TEzxH8Vv9BWOC8vx1EJN8Ud4P9EEU='),
('sha256', 'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'sha256:uWNxIxPNQjZpb7TEzxH8Vv9BWOC8vx1EJN8Ud4P9EEU'),
('sha256', 'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'),
('sha512', '26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'sha512:Jt89nVnaV01vjTWcsmILG4ZzchXDjEEt/uCkEKzq'
'GsQoWtDDcinKdOcVxEOXnaF9PXepfSrHnMXjlbBb+kvdMA=='),
('sha512', '26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'))
self.assertEqual(
digest.extract_digest_and_algorithm(
'sha512:Jt89nVnaV01vjTWcsmILG4ZzchXDjEEt_uCkEKzq'
'GsQoWtDDcinKdOcVxEOXnaF9PXepfSrHnMXjlbBb-kvdMA'),
('sha512', '26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'))
with self.assertRaises(ValueError):
digest.extract_digest_and_algorithm('')
with self.assertRaises(ValueError):
digest.extract_digest_and_algorithm(
'exactly_forty_chars_but_not_hex_encoded!')
# Too short (md5)
with self.assertRaises(ValueError):
digest.extract_digest_and_algorithm(
'd41d8cd98f00b204e9800998ecf8427e')
# but you can slip it in via the prefix notation!
self.assertEqual(
digest.extract_digest_and_algorithm(
'md5:1B2M2Y8AsgTpgAmY7PhCfg'),
('md5', 'd41d8cd98f00b204e9800998ecf8427e'))
def test_get_allowed_digests(self):
# start with defaults
allowed, deprecated = digest.get_allowed_digests(
''.split(), self.logger)
self.assertEqual(allowed, {'sha256', 'sha512', 'sha1'})
self.assertEqual(deprecated, {'sha1'})
warning_lines = self.logger.get_lines_for_level('warning')
expected_warning_line = (
'The following digest algorithms are allowed by default but '
'deprecated: sha1. Support will be disabled by default in a '
'future release, and later removed entirely.')
self.assertIn(expected_warning_line, warning_lines)
self.logger.clear()
# now with a subset
allowed, deprecated = digest.get_allowed_digests(
'sha1 sha256'.split(), self.logger)
self.assertEqual(allowed, {'sha256', 'sha1'})
self.assertEqual(deprecated, {'sha1'})
warning_lines = self.logger.get_lines_for_level('warning')
expected_warning_line = (
'The following digest algorithms are configured but '
'deprecated: sha1. Support will be removed in a future release.')
self.assertIn(expected_warning_line, warning_lines)
self.logger.clear()
# Now also with an unsupported digest
allowed, deprecated = digest.get_allowed_digests(
'sha1 sha256 md5'.split(), self.logger)
self.assertEqual(allowed, {'sha256', 'sha1'})
self.assertEqual(deprecated, {'sha1'})
warning_lines = self.logger.get_lines_for_level('warning')
self.assertIn(expected_warning_line, warning_lines)
expected_unsupported_warning_line = (
'The following digest algorithms are configured but not '
'supported: md5')
self.assertIn(expected_unsupported_warning_line, warning_lines)
self.logger.clear()
# Now with no deprecated digests
allowed, deprecated = digest.get_allowed_digests(
'sha256 sha512'.split(), self.logger)
self.assertEqual(allowed, {'sha256', 'sha512'})
self.assertEqual(deprecated, set())
warning_lines = self.logger.get_lines_for_level('warning')
self.assertFalse(warning_lines)
self.logger.clear()
# no valid digest
# Now also with an unsupported digest
with self.assertRaises(ValueError):
digest.get_allowed_digests(['md5'], self.logger)
warning_lines = self.logger.get_lines_for_level('warning')
self.assertIn(expected_unsupported_warning_line, warning_lines)

107
test/unit/common/test_utils.py
View File

@ -3836,113 +3836,6 @@ cluster_dfw1 = http://dfw1.host/v1/
self.assertEqual(u'abc_%EC%9D%BC%EC%98%81',
utils.quote(u'abc_\uc77c\uc601'))
def test_extract_digest_and_algorithm(self):
self.assertEqual(
utils.extract_digest_and_algorithm(
'b17f6ff8da0e251737aa9e3ee69a881e3e092e2f'),
('sha1', 'b17f6ff8da0e251737aa9e3ee69a881e3e092e2f'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'sha1:sw3eTSuFYrhJZGbDtGsrmsUFRGE='),
('sha1', 'b30dde4d2b8562b8496466c3b46b2b9ac5054461'))
# also good with '=' stripped
self.assertEqual(
utils.extract_digest_and_algorithm(
'sha1:sw3eTSuFYrhJZGbDtGsrmsUFRGE'),
('sha1', 'b30dde4d2b8562b8496466c3b46b2b9ac5054461'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'),
('sha256', 'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'sha256:uWNxIxPNQjZpb7TEzxH8Vv9BWOC8vx1EJN8Ud4P9EEU='),
('sha256', 'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'sha256:uWNxIxPNQjZpb7TEzxH8Vv9BWOC8vx1EJN8Ud4P9EEU'),
('sha256', 'b963712313cd4236696fb4c4cf11fc56'
'ff4158e0bcbf1d4424df147783fd1045'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'),
('sha512', '26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'sha512:Jt89nVnaV01vjTWcsmILG4ZzchXDjEEt/uCkEKzq'
'GsQoWtDDcinKdOcVxEOXnaF9PXepfSrHnMXjlbBb+kvdMA=='),
('sha512', '26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'))
self.assertEqual(
utils.extract_digest_and_algorithm(
'sha512:Jt89nVnaV01vjTWcsmILG4ZzchXDjEEt_uCkEKzq'
'GsQoWtDDcinKdOcVxEOXnaF9PXepfSrHnMXjlbBb-kvdMA'),
('sha512', '26df3d9d59da574d6f8d359cb2620b1b'
'86737215c38c412dfee0a410acea1ac4'
'285ad0c37229ca74e715c443979da17d'
'3d77a97d2ac79cc5e395b05bfa4bdd30'))
with self.assertRaises(ValueError):
utils.extract_digest_and_algorithm('')
with self.assertRaises(ValueError):
utils.extract_digest_and_algorithm(
'exactly_forty_chars_but_not_hex_encoded!')
# Too short (md5)
with self.assertRaises(ValueError):
utils.extract_digest_and_algorithm(
'd41d8cd98f00b204e9800998ecf8427e')
# but you can slip it in via the prefix notation!
self.assertEqual(
utils.extract_digest_and_algorithm('md5:1B2M2Y8AsgTpgAmY7PhCfg'),
('md5', 'd41d8cd98f00b204e9800998ecf8427e'))
def test_get_hmac(self):
self.assertEqual(
utils.get_hmac('GET', '/path', 1, 'abc'),
'b17f6ff8da0e251737aa9e3ee69a881e3e092e2f')
def test_get_hmac_ip_range(self):
self.assertEqual(
utils.get_hmac('GET', '/path', 1, 'abc', ip_range='127.0.0.1'),
'b30dde4d2b8562b8496466c3b46b2b9ac5054461')
def test_get_hmac_ip_range_non_binary_type(self):
self.assertEqual(
utils.get_hmac(u'GET', u'/path', 1, u'abc', ip_range=u'127.0.0.1'),
'b30dde4d2b8562b8496466c3b46b2b9ac5054461')
def test_get_hmac_digest(self):
self.assertEqual(
utils.get_hmac(u'GET', u'/path', 1, u'abc', digest='sha256'),
'64c5558394f86b042ce1e929b34907abd9d0a57f3e20cd3f93cffd83de0206a7')
self.assertEqual(
utils.get_hmac(u'GET', u'/path', 1, u'abc', digest=hashlib.sha256),
'64c5558394f86b042ce1e929b34907abd9d0a57f3e20cd3f93cffd83de0206a7')
self.assertEqual(
utils.get_hmac(u'GET', u'/path', 1, u'abc', digest='sha512'),
'7e95af818aec1b69b53fc2cb6d69456ec64ebda6c17b8fc8b7303b78acc8ca'
'14fc4aed96c1614a8e9d6ff45a6237711d8be294cda679624825d79aa6959b'
'5229')
self.assertEqual(
utils.get_hmac(u'GET', u'/path', 1, u'abc', digest=hashlib.sha512),
'7e95af818aec1b69b53fc2cb6d69456ec64ebda6c17b8fc8b7303b78acc8ca'
'14fc4aed96c1614a8e9d6ff45a6237711d8be294cda679624825d79aa6959b'
'5229')
def test_parse_override_options(self):
# When override_<thing> is passed in, it takes precedence.
opts = utils.parse_override_options(

22
test/unit/proxy/controllers/test_info.py
View File

@ -20,7 +20,7 @@ from mock import Mock
from swift.proxy.controllers import InfoController
from swift.proxy.server import Application as ProxyApp
from swift.common import utils, registry
from swift.common import registry, digest
from swift.common.swob import Request, HTTPException
from test.debug_logger import debug_logger
@ -133,7 +133,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, '')
sig = digest.get_hmac('GET', '/info', expires, '')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -149,7 +149,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -170,7 +170,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -180,7 +180,7 @@ class TestInfoController(unittest.TestCase):
self.assertEqual('200 OK', str(resp))
expires = int(time.time() + 86400)
sig = utils.get_hmac('HEAD', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('HEAD', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -196,7 +196,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('HEAD', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('HEAD', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -212,7 +212,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = 1
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -222,7 +222,7 @@ class TestInfoController(unittest.TestCase):
self.assertEqual('401 Unauthorized', str(resp))
expires = 'abc'
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -238,7 +238,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/foo', expires, 'secret-admin-key')
sig = digest.get_hmac('GET', '/foo', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -254,7 +254,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/foo', expires, 'invalid-admin-key')
sig = digest.get_hmac('GET', '/foo', expires, 'invalid-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
@ -272,7 +272,7 @@ class TestInfoController(unittest.TestCase):
registry._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
sig = digest.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(