openstack/swift
Code Issues Proposed changes

Merge "Stop using client headers for cross-middleware communication"

Browse Source
This commit is contained in:
Jenkins 2017-03-03 16:23:09 +00:00 committed by Gerrit Code Review
commit 75d6a5df02
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)