openstack/swift
Code Issues Proposed changes

Stop using client headers for cross-middleware communication

Browse Source 
Previously, Swift3 used client-facing HTTP headers to pass the S3 access
key, signature, and normalized request through the WSGI pipeline.
However, tempauth did not validate that Swift3 actually set the headers;
as a result, an attacker who has captured either a single valid S3-style
temporary URL or a single valid request through the S3 API may impersonate
the user that signed the URL or issued the request indefinitely through
the Swift API.

Now, the S3 authentication information will be taken from a separate
namespace in the WSGI environment, completely inaccessible to the
client. Specifically,

    environ['swift3.auth_details'] = {
        'access_key': <access key>,
        'signature': <signature>,
        'string_to_sign': <normalized request>,
    }

Note that tempauth is not expected to be in production use, but may have
been used as a template by other authentication middlewares to add their
own Swift3 support.

Change-Id: Ib90adcc2f059adaf203fba1c95b2154561ea7487
Related-Change: Ia3fbb4938f0daa8845cba4137a01cc43bc1a713c
This commit is contained in:
Tim Burke 2016-03-21 23:15:52 -07:00
parent 1f36b5dd16
commit f3ef616dc6
2 changed files with 21 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
swift/common/middleware/tempauth.py
View File

@ -24,7 +24,6 @@ import base64
from eventlet import Timeout
import six
from six.moves.urllib.parse import unquote
from swift.common.swob import Response, Request
from swift.common.swob import HTTPBadRequest, HTTPForbidden, HTTPNotFound, \
HTTPUnauthorized
@ -234,7 +233,7 @@ class TempAuth(object):
return self.app(env, start_response)
if env.get('PATH_INFO', '').startswith(self.auth_prefix):
return self.handle(env, start_response)
s3 = env.get('HTTP_AUTHORIZATION')
s3 = env.get('swift3.auth_details')
token = env.get('HTTP_X_AUTH_TOKEN', env.get('HTTP_X_STORAGE_TOKEN'))
service_token = env.get('HTTP_X_SERVICE_TOKEN')
if s3 or (token and token.startswith(self.reseller_prefix)):
@ -394,19 +393,21 @@ class TempAuth(object):
if expires < time():
groups = None
if env.get('HTTP_AUTHORIZATION'):
account_user, sign = \
env['HTTP_AUTHORIZATION'].split(' ')[1].rsplit(':', 1)
s3_auth_details = env.get('swift3.auth_details')
if s3_auth_details:
account_user = s3_auth_details['access_key']
signature_from_user = s3_auth_details['signature']
if account_user not in self.users:
return None
account, user = account_user.split(':', 1)
account_id = self.users[account_user]['url'].rsplit('/', 1)[-1]
path = env['PATH_INFO']
env['PATH_INFO'] = path.replace(account_user, account_id, 1)
msg = base64.urlsafe_b64decode(unquote(token))
key = self.users[account_user]['key']
s = base64.encodestring(hmac.new(key, msg, sha1).digest()).strip()
if s != sign:
valid_signature = base64.encodestring(hmac.new(
self.users[account_user]['key'],
s3_auth_details['string_to_sign'],
sha1).digest()).strip()
if signature_from_user != valid_signature:
return None
groups = self._get_user_groups(account, account_user, account_id)

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

@ -268,16 +268,19 @@ class TestAuth(unittest.TestCase):
def test_auth_with_s3_authorization(self):
local_app = FakeApp()
local_auth = auth.filter_factory(
{'user_s3_s3': 's3 .admin'})(local_app)
req = self._make_request('/v1/AUTH_s3',
headers={'X-Auth-Token': 't',
'AUTHORIZATION': 'AWS s3:s3:pass'})
{'user_s3_s3': 'secret .admin'})(local_app)
req = self._make_request('/v1/AUTH_s3', environ={
'swift3.auth_details': {
'access_key': 's3:s3',
'signature': b64encode('sig'),
'string_to_sign': 't'}})
with mock.patch('base64.urlsafe_b64decode') as msg, \
mock.patch('base64.encodestring') as sign:
msg.return_value = ''
sign.return_value = 'pass'
with mock.patch('hmac.new') as hmac:
hmac.return_value.digest.return_value = 'sig'
resp = req.get_response(local_auth)
self.assertEqual(hmac.mock_calls, [
mock.call('secret', 't', mock.ANY),
mock.call().digest()])
self.assertEqual(resp.status_int, 404)
self.assertEqual(local_app.calls, 1)