Fix the s3tokens endpoint

This was broken when issue_v2_token was removed, and no one noticed
because there are no tests.

The good news is, Swift3 is content to move toward supporting the
v3 format, so just start inheriting from the v3 controller.

Change-Id: I5d0c18121ba4bf8e33209daa48b9d87864951362
Related-Change: I7d3b583cbec9a095ab8cc20c5d6c0a6127e37068
Related-Change: I747de516ab69a47622eecbf8ab3faa34444b3ad5
(cherry picked from commit 3ec1aa4c19)
This commit is contained in:
Tim Burke 2017-02-22 14:22:04 +00:00 committed by Harry Rybacki
parent e1a94f39ed
commit 5c34cb43d3
4 changed files with 83 additions and 19 deletions

View File

@ -254,6 +254,20 @@ class Ec2ControllerCommon(object):
message=_('EC2 access key not found.'))
return self._convert_v3_to_ec2_credential(cred)
def render_token_data_response(self, token_id, token_data):
"""Render token data HTTP response.
Stash token ID into the X-Subject-Token header.
"""
status = (http_client.OK,
http_client.responses[http_client.OK])
headers = [('X-Subject-Token', token_id)]
return wsgi.render_response(body=token_data,
status=status,
headers=headers)
@dependency.requires('policy_api', 'token_provider_api')
class Ec2Controller(Ec2ControllerCommon, controller.V2Controller):
@ -374,7 +388,7 @@ class Ec2ControllerV3(Ec2ControllerCommon, controller.V3Controller):
token_id, token_data = self.token_provider_api.issue_token(
user_ref['id'], method_names, project_id=project_ref['id'])
return render_token_data_response(token_id, token_data)
return self.render_token_data_response(token_id, token_data)
@controller.protected(callback=_check_credential_owner_and_user_id_match)
def ec2_get_credential(self, request, user_id, credential_id):
@ -409,17 +423,3 @@ class Ec2ControllerV3(Ec2ControllerCommon, controller.V3Controller):
'credential_id': ref['access']}
ref.setdefault('links', {})
ref['links']['self'] = url
def render_token_data_response(token_id, token_data):
"""Render token data HTTP response.
Stash token ID into the X-Subject-Token header.
"""
headers = [('X-Subject-Token', token_id)]
return wsgi.render_response(body=token_data,
status=(http_client.OK,
http_client.responses[http_client.OK]),
headers=headers)

View File

@ -26,6 +26,7 @@ import hashlib
import hmac
import six
from six.moves import http_client
from keystone.common import extension
from keystone.common import json_home
@ -66,7 +67,7 @@ class S3Extension(wsgi.V3ExtensionRouter):
's3tokens', '1.0', 's3tokens'))
class S3Controller(controllers.Ec2Controller):
class S3Controller(controllers.Ec2ControllerV3):
def check_signature(self, creds_ref, credentials):
string_to_sign = base64.urlsafe_b64decode(str(credentials['token']))
@ -123,3 +124,12 @@ class S3Controller(controllers.Ec2Controller):
signature = hmac.new(signed, string_to_sign, hashlib.sha256)
return signature.hexdigest()
def render_token_data_response(self, token_id, token_data):
"""Render token data HTTP response.
Note: We neither want nor need to send back the token id.
"""
status = (http_client.OK,
http_client.responses[http_client.OK])
return wsgi.render_response(body=token_data, status=status)

View File

@ -12,21 +12,72 @@
# License for the specific language governing permissions and limitations
# under the License.
import base64
import hashlib
import hmac
import uuid
from six.moves import http_client
from keystone.contrib import s3
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import test_v3
class S3ContribCore(unit.TestCase):
class S3ContribCore(test_v3.RestfulTestCase):
def setUp(self):
super(S3ContribCore, self).setUp()
self.load_backends()
self.cred_blob, self.credential = unit.new_ec2_credential(
self.user['id'], self.project_id)
self.credential_api.create_credential(
self.credential['id'], self.credential)
self.controller = s3.S3Controller()
def test_good_response(self):
sts = 'string to sign' # opaque string from swift3
sig = hmac.new(self.cred_blob['secret'].encode('ascii'),
sts.encode('ascii'), hashlib.sha1).digest()
resp = self.post(
'/s3tokens',
body={'credentials': {
'access': self.cred_blob['access'],
'signature': base64.b64encode(sig).strip(),
'token': base64.b64encode(sts.encode('ascii')).strip(),
}},
expected_status=http_client.OK)
self.assertValidProjectScopedTokenResponse(resp, self.user,
forbid_token_id=True)
def test_bad_request(self):
self.post(
'/s3tokens',
body={},
expected_status=http_client.BAD_REQUEST)
self.post(
'/s3tokens',
body="not json",
expected_status=http_client.BAD_REQUEST)
self.post(
'/s3tokens',
expected_status=http_client.BAD_REQUEST)
def test_bad_response(self):
self.post(
'/s3tokens',
body={'credentials': {
'access': self.cred_blob['access'],
'signature': base64.b64encode(b'totally not the sig').strip(),
'token': base64.b64encode(b'string to sign').strip(),
}},
expected_status=http_client.UNAUTHORIZED)
def test_good_signature_v1(self):
creds_ref = {'secret':
u'b121dd41cdcc42fe9f70e572e84295aa'}

View File

@ -637,8 +637,11 @@ class RestfulTestCase(unit.SQLDriverOverrides, rest.RestfulTestCase,
msg = '%s is not a valid ISO 8601 extended format date time.' % dt
raise AssertionError(msg)
def assertValidTokenResponse(self, r, user=None):
self.assertTrue(r.headers.get('X-Subject-Token'))
def assertValidTokenResponse(self, r, user=None, forbid_token_id=False):
if forbid_token_id:
self.assertNotIn('X-Subject-Token', r.headers)
else:
self.assertTrue(r.headers.get('X-Subject-Token'))
token = r.result['token']
self.assertIsNotNone(token.get('expires_at'))