openstack/keystone
Code Issues Proposed changes

Add `type' filter for list_credentials_for_user

Browse Source 
When getting, or operating on a credential, EC2 controller needs to
specify it's a ec2 credential, avoiding conflicts with other type of
credentials.

Closes-Bug: #1506473
Change-Id: Id92fc87bf1be5448aa929224bbce4d3f7f4359b6
This commit is contained in:
Tony Wang 2015-10-15 17:00:22 +08:00 committed by Tony Wang
parent 44d0c2f5a5
commit 9fc1cc67d9
6 changed files with 153 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
keystone/contrib/ec2/controllers.py
View File

@ -17,7 +17,7 @@
This service allows the creation of access/secret credentials used for
the ec2 interop layer of OpenStack.
A user can create as many access/secret pairs, each of which map to a
A user can create as many access/secret pairs, each of which is mapped to a
specific project. This is required because OpenStack supports a user
belonging to multiple projects, whereas the signatures created on ec2-style
requests don't allow specification of which project the user wishes to act
@ -47,6 +47,8 @@ from keystone.common import wsgi
from keystone import exception
from keystone.i18n import _
CRED_TYPE_EC2 = 'ec2'
@dependency.requires('assignment_api', 'catalog_api', 'credential_api',
'identity_api', 'resource_api', 'role_api',
@ -184,7 +186,7 @@ class Ec2ControllerCommon(object):
'project_id': tenant_id,
'blob': jsonutils.dumps(blob),
'id': credential_id,
'type': 'ec2'}
'type': CRED_TYPE_EC2}
self.credential_api.create_credential(credential_id, cred_ref)
return {'credential': self._convert_v3_to_ec2_credential(cred_ref)}
@ -196,7 +198,7 @@ class Ec2ControllerCommon(object):
"""
self.identity_api.get_user(user_id)
credential_refs = self.credential_api.list_credentials_for_user(
user_id)
user_id, type=CRED_TYPE_EC2)
return {'credentials':
[self._convert_v3_to_ec2_credential(credential)
for credential in credential_refs]}
@ -248,14 +250,15 @@ class Ec2ControllerCommon(object):
:param credential_id: id of credential
:raises keystone.exception.Unauthorized: when credential id is invalid
or when the credential type is not ec2
:returns: credential: dict of ec2 credential.
"""
ec2_credential_id = utils.hash_access_key(credential_id)
creds = self.credential_api.get_credential(ec2_credential_id)
if not creds:
cred = self.credential_api.get_credential(ec2_credential_id)
if not cred or cred['type'] != CRED_TYPE_EC2:
raise exception.Unauthorized(
message=_('EC2 access key not found.'))
return self._convert_v3_to_ec2_credential(creds)
return self._convert_v3_to_ec2_credential(cred)
@dependency.requires('policy_api', 'token_provider_api')

7
keystone/credential/backends/sql.py
View File

@ -50,10 +50,13 @@ class Credential(credential.CredentialDriverV8):
credentials, hints)
return [s.to_dict() for s in credentials]
def list_credentials_for_user(self, user_id):
def list_credentials_for_user(self, user_id, type=None):
session = sql.get_session()
query = session.query(CredentialModel)
refs = query.filter_by(user_id=user_id).all()
query = query.filter_by(user_id=user_id)
if type:
query = query.filter_by(type=type)
refs = query.all()
return [ref.to_dict() for ref in refs]
def _get_credential(self, session, credential_id):

3
keystone/credential/core.py
View File

@ -77,10 +77,11 @@ class CredentialDriverV8(object):
raise exception.NotImplemented() # pragma: no cover
@abc.abstractmethod
def list_credentials_for_user(self, user_id):
def list_credentials_for_user(self, user_id, type=None):
"""List credentials for a user.
:param user_id: ID of a user to filter credentials by.
:param type: type of credentials to filter on.
:returns: a list of credential_refs or an empty list.

9
keystone/tests/unit/test_backend_sql.py
View File

@ -746,7 +746,8 @@ class SqlCredential(SqlTests):
def _create_credential_with_user_id(self, user_id=uuid.uuid4().hex):
credential = unit.new_credential_ref(user_id=user_id,
extra=uuid.uuid4().hex)
extra=uuid.uuid4().hex,
type=uuid.uuid4().hex)
self.credential_api.create_credential(credential['id'], credential)
return credential
@ -782,3 +783,9 @@ class SqlCredential(SqlTests):
credentials = self.credential_api.list_credentials_for_user(
self.user_foo['id'])
self._validateCredentialList(credentials, self.user_credentials)
def test_list_credentials_for_user_and_type(self):
cred = self.user_credentials[0]
credentials = self.credential_api.list_credentials_for_user(
self.user_foo['id'], type=cred['type'])
self._validateCredentialList(credentials, [cred])

94
keystone/tests/unit/test_credential.py Normal file
View File

@ -0,0 +1,94 @@
# Copyright 2015 UnitedStack, Inc
#
# 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.
import uuid
from six.moves import http_client
from keystone.common import utils
from keystone.contrib.ec2 import controllers
from keystone.tests import unit
from keystone.tests.unit import rest
CRED_TYPE_EC2 = controllers.CRED_TYPE_EC2
class TestCredentialEc2(rest.RestfulTestCase):
def setUp(self):
super(TestCredentialEc2, self).setUp()
self.user_id = self.user_foo['id']
self.project_id = self.tenant_bar['id']
def _get_token_id(self, r):
return r.result['access']['token']['id']
def _get_ec2_cred(self):
uri = self._get_ec2_cred_uri()
r = self.public_request(method='POST', token=self.get_scoped_token(),
path=uri, body={'tenant_id': self.project_id})
return r.result['credential']
def _get_ec2_cred_uri(self):
return '/v2.0/users/%s/credentials/OS-EC2' % self.user_id
def test_ec2_cannot_get_non_ec2_credential(self):
access_key = uuid.uuid4().hex
cred_id = utils.hash_access_key(access_key)
non_ec2_cred = unit.new_credential_ref(
user_id=self.user_id,
project_id=self.project_id)
non_ec2_cred['id'] = cred_id
self.credential_api.create_credential(cred_id, non_ec2_cred)
# if access_key is not found, ec2 controller raises Unauthorized
# exception
path = '/'.join([self._get_ec2_cred_uri(), access_key])
self.public_request(method='GET', token=self.get_scoped_token(),
path=path,
expected_status=http_client.UNAUTHORIZED)
def assertValidErrorResponse(self, r):
# FIXME(wwwjfy): it's copied from test_v3.py. The logic of this method
# in test_v2.py and test_v3.py (both are inherited from rest.py) has no
# difference, so they should be refactored into one place. Also, the
# function signatures in both files don't match the one in the parent
# class in rest.py.
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_ec2_list_credentials(self):
self._get_ec2_cred()
uri = self._get_ec2_cred_uri()
r = self.public_request(method='GET', token=self.get_scoped_token(),
path=uri)
cred_list = r.result['credentials']
self.assertEqual(1, len(cred_list))
# non-EC2 credentials won't be fetched
non_ec2_cred = unit.new_credential_ref(
user_id=self.user_id,
project_id=self.project_id)
non_ec2_cred['type'] = uuid.uuid4().hex
self.credential_api.create_credential(non_ec2_cred['id'],
non_ec2_cred)
r = self.public_request(method='GET', token=self.get_scoped_token(),
path=uri)
cred_list_2 = r.result['credentials']
# still one element because non-EC2 credentials are not returned.
self.assertEqual(1, len(cred_list_2))
self.assertEqual(cred_list[0], cred_list_2[0])

41
keystone/tests/unit/test_v3_credential.py
View File

@ -21,12 +21,15 @@ from oslo_config import cfg
from six.moves import http_client
from testtools import matchers
from keystone.common import utils
from keystone.contrib.ec2 import controllers
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import test_v3
CONF = cfg.CONF
CRED_TYPE_EC2 = controllers.CRED_TYPE_EC2
class CredentialBaseTestCase(test_v3.RestfulTestCase):
@ -97,7 +100,7 @@ class CredentialTestCase(CredentialBaseTestCase):
# because the type must be in the list of supported types
ec2_credential = unit.new_credential_ref(user_id=uuid.uuid4().hex,
project_id=self.project_id,
type='ec2')
type=CRED_TYPE_EC2)
ec2_resp = self.credential_api.create_credential(
ec2_credential['id'], ec2_credential)
@ -115,7 +118,7 @@ class CredentialTestCase(CredentialBaseTestCase):
cred_ec2 = r_ec2.result['credentials'][0]
self.assertValidCredentialListResponse(r_ec2, ref=ec2_resp)
self.assertEqual('ec2', cred_ec2['type'])
self.assertEqual(CRED_TYPE_EC2, cred_ec2['type'])
self.assertEqual(ec2_credential['id'], cred_ec2['id'])
def test_list_credentials_filtered_by_type_and_user_id(self):
@ -125,7 +128,7 @@ class CredentialTestCase(CredentialBaseTestCase):
# Creating credentials for two different users
credential_user1_ec2 = unit.new_credential_ref(user_id=user1_id,
type='ec2')
type=CRED_TYPE_EC2)
credential_user1_cert = unit.new_credential_ref(user_id=user1_id)
credential_user2_cert = unit.new_credential_ref(user_id=user2_id)
@ -140,7 +143,7 @@ class CredentialTestCase(CredentialBaseTestCase):
self.assertValidCredentialListResponse(r, ref=credential_user1_ec2)
self.assertThat(r.result['credentials'], matchers.HasLength(1))
cred = r.result['credentials'][0]
self.assertEqual('ec2', cred['type'])
self.assertEqual(CRED_TYPE_EC2, cred['type'])
self.assertEqual(user1_id, cred['user_id'])
def test_create_credential(self):
@ -250,7 +253,7 @@ class CredentialTestCase(CredentialBaseTestCase):
ref = unit.new_credential_ref(user_id=self.user['id'],
project_id=self.project_id,
blob='{"abc":"def"d}',
type='ec2')
type=CRED_TYPE_EC2)
# Assert bad request status when request contains invalid blob
response = self.post(
'/credentials',
@ -417,6 +420,19 @@ class TestCredentialEc2(CredentialBaseTestCase):
self.assertThat(ec2_cred['links']['self'],
matchers.EndsWith(uri))
def test_ec2_cannot_get_non_ec2_credential(self):
access_key = uuid.uuid4().hex
cred_id = utils.hash_access_key(access_key)
non_ec2_cred = unit.new_credential_ref(
user_id=self.user_id,
project_id=self.project_id)
non_ec2_cred['id'] = cred_id
self.credential_api.create_credential(cred_id, non_ec2_cred)
uri = '/'.join([self._get_ec2_cred_uri(), access_key])
# if access_key is not found, ec2 controller raises Unauthorized
# exception
self.get(uri, expected_status=http_client.UNAUTHORIZED)
def test_ec2_list_credentials(self):
"""Test ec2 credential listing."""
self._get_ec2_cred()
@ -427,13 +443,26 @@ class TestCredentialEc2(CredentialBaseTestCase):
self.assertThat(r.result['links']['self'],
matchers.EndsWith(uri))
# non-EC2 credentials won't be fetched
non_ec2_cred = unit.new_credential_ref(
user_id=self.user_id,
project_id=self.project_id)
non_ec2_cred['type'] = uuid.uuid4().hex
self.credential_api.create_credential(non_ec2_cred['id'],
non_ec2_cred)
r = self.get(uri)
cred_list_2 = r.result['credentials']
# still one element because non-EC2 credentials are not returned.
self.assertEqual(1, len(cred_list_2))
self.assertEqual(cred_list[0], cred_list_2[0])
def test_ec2_delete_credential(self):
"""Test ec2 credential deletion."""
ec2_cred = self._get_ec2_cred()
uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
cred_from_credential_api = (
self.credential_api
.list_credentials_for_user(self.user_id))
.list_credentials_for_user(self.user_id, type=CRED_TYPE_EC2))
self.assertEqual(1, len(cred_from_credential_api))
self.delete(uri)
self.assertRaises(exception.CredentialNotFound,