Fix ec2tokens validation in v2 after regression in metadata_ref removal

Since the last patch in the ocata release that removed the metadata_ref,
the ec2tokens api is broken due to unable to unpack the result of the
authenticate command (4 elements) while expecting to expand it into 5.

Change-Id: I71c4b51444ea9f7a3016b68d7dee9a4747e9c0fd
Closes-Bug: #1691111
Closes-Bug: #1635389
(cherry picked from commit 820d9d9a84)
This commit is contained in:
Jose Castro Leon 2017-05-17 14:00:34 +02:00 committed by Christian Schwede
parent 16d8f0d11f
commit e1a94f39ed
2 changed files with 208 additions and 15 deletions

View File

@ -47,6 +47,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'
@ -259,22 +260,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

View 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)