Merge "Fix ec2tokens validation in v2 after regression in metadata_ref removal"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
630d9b58fd
@ -48,6 +48,7 @@ from keystone.common import utils
|
||||
from keystone.common import wsgi
|
||||
from keystone import exception
|
||||
from keystone.i18n import _
|
||||
from keystone.token import controllers as token_controllers
|
||||
|
||||
CRED_TYPE_EC2 = 'ec2'
|
||||
|
||||
@ -274,22 +275,17 @@ class Ec2Controller(Ec2ControllerCommon, controller.V2Controller):
|
||||
|
||||
@controller.v2_ec2_deprecated
|
||||
def authenticate(self, request, credentials=None, ec2Credentials=None):
|
||||
(user_ref, tenant_ref, metadata_ref, roles_ref,
|
||||
catalog_ref) = self._authenticate(credentials=credentials,
|
||||
ec2credentials=ec2Credentials)
|
||||
(user_ref, project_ref, roles_ref, catalog_ref) = self._authenticate(
|
||||
credentials=credentials, ec2credentials=ec2Credentials
|
||||
)
|
||||
|
||||
# NOTE(morganfainberg): Make sure the data is in correct form since it
|
||||
# might be consumed external to Keystone and this is a v2.0 controller.
|
||||
# The token provider does not explicitly care about user_ref version
|
||||
# in this case, but the data is stored in the token itself and should
|
||||
# match the version
|
||||
user_ref = self.v3_to_v2_user(user_ref)
|
||||
auth_token_data = dict(user=user_ref,
|
||||
tenant=tenant_ref,
|
||||
metadata=metadata_ref,
|
||||
id='placeholder')
|
||||
(token_id, token_data) = self.token_provider_api.issue_v2_token(
|
||||
auth_token_data, roles_ref, catalog_ref)
|
||||
method_names = ['ec2credential']
|
||||
|
||||
token_id, token_data = self.token_provider_api.issue_token(
|
||||
user_ref['id'], method_names, project_id=project_ref['id'])
|
||||
|
||||
v2_helper = token_controllers.V2TokenDataHelper()
|
||||
token_data = v2_helper.v3_to_v2_token(token_data, token_id)
|
||||
return token_data
|
||||
|
||||
@controller.v2_ec2_deprecated
|
||||
|
||||
197
keystone/tests/unit/test_contrib_ec2_core.py
Normal file
197
keystone/tests/unit/test_contrib_ec2_core.py
Normal file
@ -0,0 +1,197 @@
|
||||
# Copyright 2012 OpenStack Foundation
|
||||
#
|
||||
# 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.
|
||||
|
||||
from keystoneclient.contrib.ec2 import utils as ec2_utils
|
||||
from six.moves import http_client
|
||||
|
||||
from keystone.contrib.ec2 import controllers
|
||||
from keystone.tests import unit
|
||||
from keystone.tests.unit import test_v2
|
||||
from keystone.tests.unit import test_v3
|
||||
|
||||
|
||||
class EC2ContribCoreV2(test_v2.RestfulTestCase):
|
||||
def config_overrides(self):
|
||||
super(EC2ContribCoreV2, self).config_overrides()
|
||||
|
||||
def assertValidAuthenticationResponse(self, r):
|
||||
self.assertIsNotNone(r.result.get('access'))
|
||||
self.assertIsNotNone(r.result['access'].get('token'))
|
||||
self.assertIsNotNone(r.result['access'].get('user'))
|
||||
|
||||
# validate token
|
||||
self.assertIsNotNone(r.result['access']['token'].get('id'))
|
||||
self.assertIsNotNone(r.result['access']['token'].get('expires'))
|
||||
tenant = r.result['access']['token'].get('tenant')
|
||||
if tenant is not None:
|
||||
# validate tenant
|
||||
self.assertIsNotNone(tenant.get('id'))
|
||||
self.assertIsNotNone(tenant.get('name'))
|
||||
|
||||
# validate user
|
||||
self.assertIsNotNone(r.result['access']['user'].get('id'))
|
||||
self.assertIsNotNone(r.result['access']['user'].get('name'))
|
||||
|
||||
def assertValidErrorResponse(self, r):
|
||||
resp = r.result
|
||||
self.assertIsNotNone(resp.get('error'))
|
||||
self.assertIsNotNone(resp['error'].get('code'))
|
||||
self.assertIsNotNone(resp['error'].get('title'))
|
||||
self.assertIsNotNone(resp['error'].get('message'))
|
||||
self.assertEqual(int(resp['error']['code']), r.status_code)
|
||||
|
||||
def test_valid_authentication_response_with_proper_secret(self):
|
||||
cred_blob, credential = unit.new_ec2_credential(
|
||||
self.user_foo['id'], self.tenant_bar['id'])
|
||||
|
||||
self.credential_api.create_credential(
|
||||
credential['id'], credential)
|
||||
|
||||
signer = ec2_utils.Ec2Signer(cred_blob['secret'])
|
||||
credentials = {
|
||||
'access': cred_blob['access'],
|
||||
'secret': cred_blob['secret'],
|
||||
'host': 'localhost',
|
||||
'verb': 'GET',
|
||||
'path': '/',
|
||||
'params': {
|
||||
'SignatureVersion': '2',
|
||||
'Action': 'Test',
|
||||
'Timestamp': '2007-01-31T23:59:59Z'
|
||||
},
|
||||
}
|
||||
credentials['signature'] = signer.generate(credentials)
|
||||
resp = self.public_request(
|
||||
method='POST',
|
||||
path='/v2.0/ec2tokens',
|
||||
body={'credentials': credentials},
|
||||
expected_status=http_client.OK)
|
||||
self.assertValidAuthenticationResponse(resp)
|
||||
|
||||
def test_authenticate_with_empty_body_returns_bad_request(self):
|
||||
self.public_request(
|
||||
method='POST',
|
||||
path='/v2.0/ec2tokens',
|
||||
body={},
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_authenticate_without_json_request_returns_bad_request(self):
|
||||
self.public_request(
|
||||
method='POST',
|
||||
path='/v2.0/ec2tokens',
|
||||
body='not json',
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_authenticate_without_request_body_returns_bad_request(self):
|
||||
self.public_request(
|
||||
method='POST',
|
||||
path='/v2.0/ec2tokens',
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_authenticate_without_proper_secret_returns_unauthorized(self):
|
||||
cred_blob, credential = unit.new_ec2_credential(
|
||||
self.user_foo['id'], self.tenant_bar['id'])
|
||||
|
||||
self.credential_api.create_credential(
|
||||
credential['id'], credential)
|
||||
|
||||
signer = ec2_utils.Ec2Signer('totally not the secret')
|
||||
credentials = {
|
||||
'access': cred_blob['access'],
|
||||
'secret': 'totally not the secret',
|
||||
'host': 'localhost',
|
||||
'verb': 'GET',
|
||||
'path': '/',
|
||||
'params': {
|
||||
'SignatureVersion': '2',
|
||||
'Action': 'Test',
|
||||
'Timestamp': '2007-01-31T23:59:59Z'
|
||||
},
|
||||
}
|
||||
credentials['signature'] = signer.generate(credentials)
|
||||
self.public_request(
|
||||
method='POST',
|
||||
path='/v2.0/ec2tokens',
|
||||
body={'credentials': credentials},
|
||||
expected_status=http_client.UNAUTHORIZED)
|
||||
|
||||
|
||||
class EC2ContribCoreV3(test_v3.RestfulTestCase):
|
||||
def setUp(self):
|
||||
super(EC2ContribCoreV3, self).setUp()
|
||||
|
||||
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 = controllers.Ec2ControllerV3
|
||||
|
||||
def test_valid_authentication_response_with_proper_secret(self):
|
||||
signer = ec2_utils.Ec2Signer(self.cred_blob['secret'])
|
||||
credentials = {
|
||||
'access': self.cred_blob['access'],
|
||||
'secret': self.cred_blob['secret'],
|
||||
'host': 'localhost',
|
||||
'verb': 'GET',
|
||||
'path': '/',
|
||||
'params': {
|
||||
'SignatureVersion': '2',
|
||||
'Action': 'Test',
|
||||
'Timestamp': '2007-01-31T23:59:59Z'
|
||||
},
|
||||
}
|
||||
credentials['signature'] = signer.generate(credentials)
|
||||
resp = self.post(
|
||||
'/ec2tokens',
|
||||
body={'credentials': credentials},
|
||||
expected_status=http_client.OK)
|
||||
self.assertValidProjectScopedTokenResponse(resp, self.user)
|
||||
|
||||
def test_authenticate_with_empty_body_returns_bad_request(self):
|
||||
self.post(
|
||||
'/ec2tokens',
|
||||
body={},
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_authenticate_without_json_request_returns_bad_request(self):
|
||||
self.post(
|
||||
'/ec2tokens',
|
||||
body='not json',
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_authenticate_without_request_body_returns_bad_request(self):
|
||||
self.post(
|
||||
'/ec2tokens',
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_authenticate_without_proper_secret_returns_unauthorized(self):
|
||||
signer = ec2_utils.Ec2Signer('totally not the secret')
|
||||
credentials = {
|
||||
'access': self.cred_blob['access'],
|
||||
'secret': 'totally not the secret',
|
||||
'host': 'localhost',
|
||||
'verb': 'GET',
|
||||
'path': '/',
|
||||
'params': {
|
||||
'SignatureVersion': '2',
|
||||
'Action': 'Test',
|
||||
'Timestamp': '2007-01-31T23:59:59Z'
|
||||
},
|
||||
}
|
||||
credentials['signature'] = signer.generate(credentials)
|
||||
self.post(
|
||||
'/ec2tokens',
|
||||
body={'credentials': credentials},
|
||||
expected_status=http_client.UNAUTHORIZED)
|
||||
Loading…
Reference in New Issue
Block a user