openstack/keystone
Code Issues Proposed changes

Fix security issues with EC2 credentials

Browse Source 
This change addresses several issues in the creation and use of EC2/S3
credentials with keystone tokens.

1. Disable altering credential owner attributes or metadata

Without this patch, an authenticated user can create an EC2 credential
for themself for a project they have a role on, then update the
credential to target a user and project completely unrelated to them. In
the worst case, this could be the admin user and a project the admin
user has a role assignment on. A token granted for an altered credential
like this would allow the user to masquerade as the victim user. This
patch ensures that when updating a credential, the new form of the
credential is one the acting user has access to: if the system admin
user is changing the credential, the new user ID or project ID could be
anything, but regular users may only change the credential to be one
that they still own.

Relatedly, when a user uses an application credential or a trust to
create an EC2 credential, keystone automatically adds the trust ID or
application credential ID as metadata in the EC2 access blob so that it
knows how the token can be scoped when it is used. Without this patch, a
user who has created a credential in this way can update the access blob
to remove or alter this metadata and escalate their privileges to be
fully authorized for the trustor's, application credential creator's, or
OAuth1 access token authorizor's privileges on the project. This patch
fixes the issue by simply disallowing updates to keystone-controlled
metadata in the credential.

2. Respect token roles when creating EC2 credentials

Without this patch, a trustee, an application credential user, or an
OAuth1 access token holder could create an EC2 credential or an
application credential using any roles the trustor, application
credential creator, or access token authorizor had on the project,
regardless of whether the creator had delegated only a limited subset of
roles. This was because the trust_id attribute of the EC2 access blob
was ignored, and no metadata for the application credential or access
token was recorded either. This change ensures that the access
delegation resource is recorded in the metadata of the EC2 credential
when created and passed to the token provider when used for
authentication so that the token provider can look up the correct roles
for the request.

Conflicts (six removal in  e2d83ae9, pep8 fixes in e2d83ae9):
      keystone/api/credentials.py
      keystone/tests/unit/test_v3_application_credential.py
      keystone/tests/unit/test_v3_credential.py

Conflicts due to flask reorg:
	keystone/api/_shared/EC2_S3_Resource.py
	keystone/api/credentials.py
	keystone/api/users.py
	keystone/tests/unit/test_v3_credential.py

Conflicts due to token provider refactor between Queens and Rocky. Also
conflicted due to changing method signature for _authenticate().
    keystone/credential/controllers.py
    keystone/contrib/ec2/controllers.py
    keystone/application_credential/controllers.py
    keystone/token/providers/common.py

Moved the test_update_credential_non_owner unit test to
CredentialSelfServiceTestCase since in this branch the default policies
are not affected by #1872733.

Change-Id: I39d0d705839fbe31ac518ac9a82959e108cb7c1d
Closes-bug: #1872733
Closes-bug: #1872755
Closes-bug: #1872735
(cherry picked from commit 37e9907a17)
(cherry picked from commit 2f2736ebb2)
(cherry picked from commit 27caafe3daa552663719954f2cd6713dd4493178)
(cherry picked from commit bfba75fc3c5c8f119f74dbf31347e008824a2134)
(cherry picked from commit 53d1ccb8a1)
This commit is contained in:
Colleen Murphy 2020-04-14 16:47:44 -07:00 committed by Lance Bragstad
parent fac58fdfbe
commit 487c7276c7
9 changed files with 707 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
keystone/application_credential/controllers.py
View File

@ -60,6 +60,25 @@ class ApplicationCredentialV3(controller.V3Controller):
role['name']))
return roles
def _get_roles(self, app_cred_data, token):
if app_cred_data.get('roles'):
roles = self._normalize_role_list(app_cred_data['roles'])
# NOTE(cmurphy): The user is not allowed to add a role that is not
# in their token. This is to prevent trustees or application
# credential users from escallating their privileges to include
# additional roles that the trustor or application credential
# creator has assigned on the project.
token_roles = [r['id'] for r in token['roles']]
for role in roles:
if role['id'] not in token_roles:
detail = _('Cannot create an application credential with '
'unassigned role')
raise exception.ApplicationCredentialValidationError(
detail=detail)
else:
roles = token['roles']
return roles
def _generate_secret(self):
length = 64
secret = os.urandom(length)
@ -111,8 +130,7 @@ class ApplicationCredentialV3(controller.V3Controller):
app_cred['secret'] = self._generate_secret()
app_cred['user_id'] = user_id
app_cred['project_id'] = project_id
app_cred['roles'] = self._normalize_role_list(
app_cred.get('roles', token['roles']))
app_cred['roles'] = self._get_roles(app_cred, token)
if app_cred.get('expires_at'):
app_cred['expires_at'] = utils.parse_expiration_date(
app_cred['expires_at'])

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

@ -311,7 +311,8 @@ class Ec2ControllerCommon(provider_api.ProviderAPIMixin, object):
def _authenticate(self, credentials=None, ec2credentials=None):
"""Common code shared between the V2 and V3 authenticate methods.
:returns: user_ref, tenant_ref, roles_ref, catalog_ref
:returns: user_ref, tenant_ref, roles_ref, trust_ref, app_cred_ref,
auth_context
"""
# FIXME(ja): validate that a service token was used!
@ -344,18 +345,44 @@ class Ec2ControllerCommon(provider_api.ProviderAPIMixin, object):
sys.exc_info()[2])
self._check_timestamp(credentials)
roles = self.assignment_api.get_roles_for_user_and_project(
user_ref['id'], tenant_ref['id']
)
trustee_user_id = None
auth_context = None
trust_ref = {}
app_cred_ref = {}
if creds_ref['trust_id']:
trust_ref = PROVIDERS.trust_api.get_trust(creds_ref['trust_id'])
roles = [r['id'] for r in trust_ref['roles']]
# NOTE(cmurphy): if this credential was created using a
# trust-scoped token with impersonation, the user_id will be for
# the trustor, not the trustee. In this case, issuing a
# trust-scoped token to the trustor will fail. In order to get a
# trust-scoped token, use the user ID of the trustee. With
# impersonation, the resulting token will still be for the trustor.
# Without impersonation, the token will be for the trustee.
if trust_ref['impersonation'] is True:
trustee_user_id = trust_ref['trustee_user_id']
user_ref = self.identity_api.get_user(trustee_user_id)
elif creds_ref['app_cred_id']:
ac_client = PROVIDERS.application_credential_api
app_cred_ref = ac_client.get_application_credential(
creds_ref['app_cred_id'])
roles = [r['id'] for r in app_cred_ref['roles']]
elif creds_ref['access_token_id']:
access_token = PROVIDERS.oauth_api.get_access_token(
creds_ref['access_token_id'])
roles = jsonutils.loads(access_token['role_ids'])
auth_context = {'access_token_id': creds_ref['access_token_id']}
else:
roles = PROVIDERS.assignment_api.get_roles_for_user_and_project(
user_ref['id'], tenant_ref['id'])
if not roles:
raise exception.Unauthorized(
message=_('User not valid for tenant.'))
roles_ref = [self.role_api.get_role(role_id) for role_id in roles]
catalog_ref = self.catalog_api.get_catalog(
user_ref['id'], tenant_ref['id'])
return user_ref, tenant_ref, roles_ref, catalog_ref
return (user_ref, tenant_ref, roles_ref, trust_ref, app_cred_ref,
auth_context)
def create_credential(self, request, user_id, tenant_id):
"""Create a secret/access pair for use with ec2 style auth.
@ -435,7 +462,9 @@ class Ec2ControllerCommon(provider_api.ProviderAPIMixin, object):
'tenant_id': credential.get('project_id'),
'access': blob.get('access'),
'secret': blob.get('secret'),
'trust_id': blob.get('trust_id')}
'trust_id': blob.get('trust_id'),
'app_cred_id': blob.get('app_cred_id'),
'access_token_id': blob.get('access_token_id')}
def _get_credentials(self, credential_id):
"""Return credentials from an ID.
@ -471,7 +500,8 @@ class Ec2Controller(Ec2ControllerCommon, controller.V2Controller):
@controller.v2_ec2_deprecated
def authenticate(self, request, credentials=None, ec2Credentials=None):
(user_ref, project_ref, roles_ref, catalog_ref) = self._authenticate(
(user_ref, project_ref, roles_ref, trust_ref, app_cred_ref,
auth_context) = self._authenticate(
credentials=credentials, ec2credentials=ec2Credentials
)
@ -576,14 +606,17 @@ class Ec2ControllerV3(Ec2ControllerCommon, controller.V3Controller):
self.check_protection(request, prep_info, ref)
def authenticate(self, context, credentials=None, ec2Credentials=None):
(user_ref, project_ref, roles_ref, catalog_ref) = self._authenticate(
credentials=credentials, ec2credentials=ec2Credentials
)
(user_ref, project_ref, roles_ref, trust_ref, app_cred_ref,
auth_context) = (self._authenticate(
credentials=credentials, ec2credentials=ec2Credentials))
method_names = ['ec2credential']
token_id, token_data = self.token_provider_api.issue_token(
user_ref['id'], method_names, project_id=project_ref['id'])
user_ref['id'], method_names, project_id=project_ref['id'],
trust=trust_ref, app_cred_id=app_cred_ref.get('id'),
auth_context=auth_context
)
return self.render_token_data_response(token_id, token_data)
@controller.protected(callback=_check_credential_owner_and_user_id_match)

76
keystone/credential/controllers.py
View File

@ -13,6 +13,7 @@
# under the License.
import hashlib
import six
from oslo_serialization import jsonutils
@ -35,30 +36,41 @@ class CredentialV3(controller.V3Controller):
super(CredentialV3, self).__init__()
self.get_member_from_driver = PROVIDERS.credential_api.get_credential
def _assign_unique_id(self, ref, trust_id=None):
# Generates and assigns a unique identifier to
# a credential reference.
def _validate_blob_json(self, ref):
try:
blob = jsonutils.loads(ref.get('blob'))
except (ValueError, TabError):
raise exception.ValidationError(
message=_('Invalid blob in credential'))
if not blob or not isinstance(blob, dict):
raise exception.ValidationError(attribute='blob',
target='credential')
if blob.get('access') is None:
raise exception.ValidationError(attribute='access',
target='credential')
return blob
def _assign_unique_id(
self, ref, trust_id=None, app_cred_id=None, access_token_id=None):
# Generates and assigns a unique identifier to a credential reference.
if ref.get('type', '').lower() == 'ec2':
try:
blob = jsonutils.loads(ref.get('blob'))
except (ValueError, TypeError):
raise exception.ValidationError(
message=_('Invalid blob in credential'))
if not blob or not isinstance(blob, dict):
raise exception.ValidationError(attribute='blob',
target='credential')
if blob.get('access') is None:
raise exception.ValidationError(attribute='access',
target='blob')
blob = self._validate_blob_json(ref)
ret_ref = ref.copy()
ret_ref['id'] = hashlib.sha256(
blob['access'].encode('utf8')).hexdigest()
# Update the blob with the trust_id, so credentials created
# with a trust scoped token will result in trust scoped
# tokens when authentication via ec2tokens happens
# update the blob with the trust_id, so credentials
# created with a trust- token will result in
# trust- cred-scoped tokens when authentication via
# ec2tokens happens
if trust_id is not None:
blob['trust_id'] = trust_id
ret_ref['blob'] = jsonutils.dumps(blob)
if app_cred_id is not None:
blob['app_cred_id'] = app_cred_id
ret_ref['blob'] = jsonutils.dumps(blob)
if access_token_id is not None:
blob['access_token_id'] = access_token_id
ret_ref['blob'] = jsonutils.dumps(blob)
return ret_ref
else:
return super(CredentialV3, self)._assign_unique_id(ref)
@ -66,8 +78,14 @@ class CredentialV3(controller.V3Controller):
@controller.protected()
def create_credential(self, request, credential):
validation.lazy_validate(schema.credential_create, credential)
trust_id = request.context.trust_id
access_token_id = request.context.oauth_access_token_id
app_cred_id = request.auth_context['token'].get(
'application_credential', {}).get('id')
ref = self._assign_unique_id(self._normalize_dict(credential),
request.context.trust_id)
trust_id=trust_id,
app_cred_id=app_cred_id,
access_token_id=access_token_id)
ref = PROVIDERS.credential_api.create_credential(ref['id'], ref)
return CredentialV3.wrap_member(request.context_dict, ref)
@ -97,11 +115,31 @@ class CredentialV3(controller.V3Controller):
ret_ref = self._blob_to_json(ref)
return CredentialV3.wrap_member(request.context_dict, ret_ref)
def _validate_blob_update_keys(self, credential, ref):
if credential.get('type', '').lower() == 'ec2':
new_blob = self._validate_blob_json(ref)
old_blob = credential.get('blob')
if isinstance(old_blob, six.string_types):
old_blob = jsonutils.loads(old_blob)
# if there was a scope set, prevent changing it or unsetting it
for key in ['trust_id', 'app_cred_id', 'access_token_id']:
if old_blob.get(key) != new_blob.get(key):
message = _('%s can not be updated for credential') % key
raise exception.ValidationError(message=message)
@controller.protected()
def update_credential(self, request, credential_id, credential):
current = self.credential_api.get_credential(credential_id)
validation.lazy_validate(schema.credential_update, credential)
self._validate_blob_update_keys(current.copy(), credential.copy())
self._require_matching_id(credential_id, credential)
# Check that the user hasn't illegally modified the owner or scope
target = {'credential': dict(current, **credential)}
prep_info = {'f_name': 'update_credential',
'input_attr': {}}
self.check_protection(
request, prep_info, target_attr=target
)
ref = PROVIDERS.credential_api.update_credential(
credential_id, credential
)

30
keystone/tests/unit/test_v3_application_credential.py
View File

@ -139,6 +139,36 @@ class ApplicationCredentialTestCase(test_v3.RestfulTestCase):
token=token_data.headers['x-subject-token'],
expected_status=http_client.FORBIDDEN)
def test_create_application_credential_with_trust(self):
second_role = unit.new_role_ref(name='reader')
PROVIDERS.role_api.create_role(second_role['id'], second_role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_id, self.project_id, second_role['id'])
pw_token = self.get_scoped_token()
# create a self-trust - only the roles are important for this test
trust_ref = unit.new_trust_ref(
trustor_user_id=self.user_id,
trustee_user_id=self.user_id,
project_id=self.project_id,
role_ids=[second_role['id']])
resp = self.post('/OS-TRUST/trusts',
token=pw_token,
body={'trust': trust_ref})
trust_id = resp.json['trust']['id']
trust_auth = self.build_authentication_request(
user_id=self.user_id,
password=self.user['password'],
trust_id=trust_id)
trust_token = self.v3_create_token(
trust_auth).headers['X-Subject-Token']
app_cred = self._app_cred_body(roles=[{'id': self.role_id}])
# only the roles from the trust token should be allowed, even if
# the user has the role assigned on the project
self.post('/users/%s/application_credentials' % self.user_id,
token=trust_token,
body=app_cred,
expected_status=http_client.BAD_REQUEST)
def test_create_application_credential_allow_recursion(self):
roles = [{'id': self.role_id}]
app_cred_body_1 = self._app_cred_body(roles=roles)

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

@ -19,7 +19,7 @@ import uuid
from keystoneclient.contrib.ec2 import utils as ec2_utils
import mock
from oslo_db import exception as oslo_db_exception
from six.moves import http_client
from six.moves import http_client, urllib
from testtools import matchers
from keystone.common import provider_api
@ -27,8 +27,10 @@ from keystone.common import utils
from keystone.contrib.ec2 import controllers
from keystone.credential.providers import fernet as credential_fernet
from keystone import exception
from keystone import oauth1
from keystone.tests import unit
from keystone.tests.unit import ksfixtures
from keystone.tests.unit.ksfixtures import temporaryfile
from keystone.tests.unit import test_v3
@ -63,6 +65,33 @@ class CredentialBaseTestCase(test_v3.RestfulTestCase):
return json.dumps(blob), credential_id
def _test_get_token(self, access, secret):
"""Test signature validation with the access/secret provided."""
signer = ec2_utils.Ec2Signer(secret)
params = {'SignatureMethod': 'HmacSHA256',
'SignatureVersion': '2',
'AWSAccessKeyId': access}
request = {'host': 'foo',
'verb': 'GET',
'path': '/bar',
'params': params}
signature = signer.generate(request)
# Now make a request to validate the signed dummy request via the
# ec2tokens API. This proves the v3 ec2 credentials actually work.
sig_ref = {'access': access,
'signature': signature,
'host': 'foo',
'verb': 'GET',
'path': '/bar',
'params': params}
r = self.post(
'/ec2tokens',
body={'ec2Credentials': sig_ref},
expected_status=http_client.OK)
self.assertValidTokenResponse(r)
return r.result['token']
class CredentialTestCase(CredentialBaseTestCase):
"""Test credential CRUD."""
@ -246,6 +275,90 @@ class CredentialTestCase(CredentialBaseTestCase):
'credential_id': credential_id},
body={'credential': update_ref})
def test_update_ec2_credential_change_trust_id(self):
"""Call ``PATCH /credentials/{credential_id}``."""
blob, ref = unit.new_ec2_credential(user_id=self.user['id'],
project_id=self.project_id)
blob['trust_id'] = uuid.uuid4().hex
ref['blob'] = json.dumps(blob)
r = self.post(
'/credentials',
body={'credential': ref})
self.assertValidCredentialResponse(r, ref)
credential_id = r.result.get('credential')['id']
# Try changing to a different trust
blob['trust_id'] = uuid.uuid4().hex
update_ref = {'blob': json.dumps(blob)}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
body={'credential': update_ref},
expected_status=http_client.BAD_REQUEST)
# Try removing the trust
del blob['trust_id']
update_ref = {'blob': json.dumps(blob)}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
body={'credential': update_ref},
expected_status=http_client.BAD_REQUEST)
def test_update_ec2_credential_change_app_cred_id(self):
"""Call ``PATCH /credentials/{credential_id}``."""
blob, ref = unit.new_ec2_credential(user_id=self.user['id'],
project_id=self.project_id)
blob['app_cred_id'] = uuid.uuid4().hex
ref['blob'] = json.dumps(blob)
r = self.post(
'/credentials',
body={'credential': ref})
self.assertValidCredentialResponse(r, ref)
credential_id = r.result.get('credential')['id']
# Try changing to a different app cred
blob['app_cred_id'] = uuid.uuid4().hex
update_ref = {'blob': json.dumps(blob)}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
body={'credential': update_ref},
expected_status=http_client.BAD_REQUEST)
# Try removing the app cred
del blob['app_cred_id']
update_ref = {'blob': json.dumps(blob)}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
body={'credential': update_ref},
expected_status=http_client.BAD_REQUEST)
def test_update_ec2_credential_change_access_token_id(self):
"""Call ``PATCH /credentials/{credential_id}``."""
blob, ref = unit.new_ec2_credential(user_id=self.user['id'],
project_id=self.project_id)
blob['access_token_id'] = uuid.uuid4().hex
ref['blob'] = json.dumps(blob)
r = self.post(
'/credentials',
body={'credential': ref})
self.assertValidCredentialResponse(r, ref)
credential_id = r.result.get('credential')['id']
# Try changing to a different access token
blob['access_token_id'] = uuid.uuid4().hex
update_ref = {'blob': json.dumps(blob)}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
body={'credential': update_ref},
expected_status=http_client.BAD_REQUEST)
# Try removing the access token
del blob['access_token_id']
update_ref = {'blob': json.dumps(blob)}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
body={'credential': update_ref},
expected_status=http_client.BAD_REQUEST)
def test_delete_credential(self):
"""Call ``DELETE /credentials/{credential_id}``."""
self.delete(
@ -381,7 +494,132 @@ class CredentialTestCase(CredentialBaseTestCase):
self.assertValidCredentialResponse(r, ref)
class TestCredentialTrustScoped(test_v3.RestfulTestCase):
class CredentialSelfServiceTestCase(CredentialBaseTestCase):
"""Test self-service credential CRUD."""
def _policy_fixture(self):
return ksfixtures.Policy(self.tmpfilename, self.config_fixture)
def _set_policy(self, new_policy):
with open(self.tmpfilename, "w") as policyfile:
policyfile.write(json.dumps(new_policy))
def setUp(self):
self.tempfile = self.useFixture(temporaryfile.SecureTempFile())
self.tmpfilename = self.tempfile.file_name
super(CredentialSelfServiceTestCase, self).setUp()
# set the self-service credential policies
self_service_credential_policies = {
"identity:create_credential": "user_id:%(credential.user_id)s",
"identity:list_credentials": "user_id:%(user_id)s",
"identity:get_credential": "user_id:%(target.credential.user_id)s",
"identity:update_credential":
"user_id:%(target.credential.user_id)s",
"identity:delete_credential":
"user_id:%(target.credential.user_id)s"
}
self._set_policy(self_service_credential_policies)
# remove the 'admin' role from user and replace it with an
# arbitrary role
PROVIDERS.assignment_api.remove_role_from_user_and_project(
self.user_id, self.project_id, self.role_id)
self.arbitrary_role = unit.new_role_ref(name=uuid.uuid4().hex)
PROVIDERS.role_api.create_role(self.arbitrary_role['id'],
self.arbitrary_role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_id, self.project_id, self.arbitrary_role['id'])
self.credential = unit.new_credential_ref(user_id=self.user['id'],
project_id=self.project_id)
PROVIDERS.credential_api.create_credential(
self.credential['id'],
self.credential)
def test_list_credentials_filtered_by_user_id(self):
"""Call ``GET /credentials?user_id={user_id}``."""
credential = unit.new_credential_ref(user_id=uuid.uuid4().hex)
PROVIDERS.credential_api.create_credential(
credential['id'], credential
)
r = self.get('/credentials?user_id=%s' % self.user['id'])
self.assertValidCredentialListResponse(r, ref=self.credential)
for cred in r.result['credentials']:
self.assertEqual(self.user['id'], cred['user_id'])
def test_create_credential(self):
"""Call ``POST /credentials``."""
ref = unit.new_credential_ref(user_id=self.user['id'])
r = self.post(
'/credentials',
body={'credential': ref})
self.assertValidCredentialResponse(r, ref)
def test_get_credential(self):
"""Call ``GET /credentials/{credential_id}``."""
r = self.get(
'/credentials/%(credential_id)s' % {
'credential_id': self.credential['id']})
self.assertValidCredentialResponse(r, self.credential)
def test_update_credential(self):
"""Call ``PATCH /credentials/{credential_id}``."""
ref = unit.new_credential_ref(user_id=self.user['id'],
project_id=self.project_id)
del ref['id']
r = self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': self.credential['id']},
body={'credential': ref})
self.assertValidCredentialResponse(r, ref)
def test_delete_credential(self):
"""Call ``DELETE /credentials/{credential_id}``."""
self.delete(
'/credentials/%(credential_id)s' % {
'credential_id': self.credential['id']})
def test_update_credential_non_owner(self):
"""Call ``PATCH /credentials/{credential_id}``."""
alt_user = unit.create_user(
PROVIDERS.identity_api, domain_id=self.domain_id)
alt_user_id = alt_user['id']
alt_project = unit.new_project_ref(domain_id=self.domain_id)
alt_project_id = alt_project['id']
PROVIDERS.resource_api.create_project(
alt_project['id'], alt_project)
alt_role = unit.new_role_ref(name='reader')
alt_role_id = alt_role['id']
PROVIDERS.role_api.create_role(alt_role_id, alt_role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
alt_user_id, alt_project_id, alt_role_id)
auth = self.build_authentication_request(
user_id=alt_user_id,
password=alt_user['password'],
project_id=alt_project_id)
ref = unit.new_credential_ref(user_id=alt_user_id,
project_id=alt_project_id)
r = self.post(
'/credentials',
auth=auth,
body={'credential': ref})
self.assertValidCredentialResponse(r, ref)
credential_id = r.result.get('credential')['id']
# Cannot change the credential to be owned by another user
update_ref = {'user_id': self.user_id, 'project_id': self.project_id}
self.patch(
'/credentials/%(credential_id)s' % {
'credential_id': credential_id},
expected_status=403,
auth=auth,
body={'credential': update_ref})
class TestCredentialTrustScoped(CredentialBaseTestCase):
"""Test credential with trust scoped token."""
def setUp(self):
@ -434,7 +672,7 @@ class TestCredentialTrustScoped(test_v3.RestfulTestCase):
token_id = r.headers.get('X-Subject-Token')
# Create the credential with the trust scoped token
blob, ref = unit.new_ec2_credential(user_id=self.user['id'],
blob, ref = unit.new_ec2_credential(user_id=self.user_id,
project_id=self.project_id)
r = self.post('/credentials', body={'credential': ref}, token=token_id)
@ -451,6 +689,21 @@ class TestCredentialTrustScoped(test_v3.RestfulTestCase):
self.assertEqual(hashlib.sha256(access).hexdigest(),
r.result['credential']['id'])
# Create a role assignment to ensure that it is ignored and only the
# trust-delegated roles are used
role = unit.new_role_ref(name='reader')
role_id = role['id']
PROVIDERS.role_api.create_role(role_id, role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_id, self.project_id, role_id)
ret_blob = json.loads(r.result['credential']['blob'])
ec2token = self._test_get_token(
access=ret_blob['access'], secret=ret_blob['secret'])
ec2_roles = [rl['id'] for rl in ec2token['roles']]
self.assertIn(self.role_id, ec2_roles)
self.assertNotIn(role_id, ec2_roles)
# Create second ec2 credential with the same access key id and check
# for conflict.
self.post(
@ -460,35 +713,232 @@ class TestCredentialTrustScoped(test_v3.RestfulTestCase):
expected_status=http_client.CONFLICT)
class TestCredentialAppCreds(CredentialBaseTestCase):
"""Test credential with application credential token."""
def setUp(self):
super(TestCredentialAppCreds, self).setUp()
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'credential',
credential_fernet.MAX_ACTIVE_KEYS
)
)
def test_app_cred_ec2_credential(self):
"""Test creating ec2 credential from an application credential.
Call ``POST /credentials``.
"""
# Create the app cred
ref = {
'name': uuid.uuid4().hex,
'roles': [{'id': self.role_id}]
}
r = self.post('/users/%s/application_credentials' % self.user_id,
body={'application_credential': ref})
app_cred = r.result['application_credential']
# Get an application credential token
auth_data = self.build_authentication_request(
app_cred_id=app_cred['id'],
secret=app_cred['secret'])
r = self.v3_create_token(auth_data)
token_id = r.headers.get('X-Subject-Token')
# Create the credential with the app cred token
blob, ref = unit.new_ec2_credential(user_id=self.user_id,
project_id=self.project_id)
r = self.post('/credentials', body={'credential': ref}, token=token_id)
# We expect the response blob to contain the app_cred_id
ret_ref = ref.copy()
ret_blob = blob.copy()
ret_blob['app_cred_id'] = app_cred['id']
ret_ref['blob'] = json.dumps(ret_blob)
self.assertValidCredentialResponse(r, ref=ret_ref)
# Assert credential id is same as hash of access key id for
# ec2 credentials
access = blob['access'].encode('utf-8')
self.assertEqual(hashlib.sha256(access).hexdigest(),
r.result['credential']['id'])
# Create a role assignment to ensure that it is ignored and only the
# roles in the app cred are used
role = unit.new_role_ref(name='reader')
role_id = role['id']
PROVIDERS.role_api.create_role(role_id, role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_id, self.project_id, role_id)
ret_blob = json.loads(r.result['credential']['blob'])
ec2token = self._test_get_token(
access=ret_blob['access'], secret=ret_blob['secret'])
ec2_roles = [rl['id'] for rl in ec2token['roles']]
self.assertIn(self.role_id, ec2_roles)
self.assertNotIn(role_id, ec2_roles)
# Create second ec2 credential with the same access key id and check
# for conflict.
self.post(
'/credentials',
body={'credential': ref},
token=token_id,
expected_status=http_client.CONFLICT)
class TestCredentialAccessToken(CredentialBaseTestCase):
"""Test credential with access token."""
def setUp(self):
super(TestCredentialAccessToken, self).setUp()
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'credential',
credential_fernet.MAX_ACTIVE_KEYS
)
)
self.base_url = 'http://localhost/v3'
def _urllib_parse_qs_text_keys(self, content):
results = urllib.parse.parse_qs(content)
return {key.decode('utf-8'): value for key, value in results.items()}
def _create_single_consumer(self):
endpoint = '/OS-OAUTH1/consumers'
ref = {'description': uuid.uuid4().hex}
resp = self.post(endpoint, body={'consumer': ref})
return resp.result['consumer']
def _create_request_token(self, consumer, project_id, base_url=None):
endpoint = '/OS-OAUTH1/request_token'
client = oauth1.Client(consumer['key'],
client_secret=consumer['secret'],
signature_method=oauth1.SIG_HMAC,
callback_uri="oob")
headers = {'requested_project_id': project_id}
if not base_url:
base_url = self.base_url
url, headers, body = client.sign(base_url + endpoint,
http_method='POST',
headers=headers)
return endpoint, headers
def _create_access_token(self, consumer, token, base_url=None):
endpoint = '/OS-OAUTH1/access_token'
client = oauth1.Client(consumer['key'],
client_secret=consumer['secret'],
resource_owner_key=token.key,
resource_owner_secret=token.secret,
signature_method=oauth1.SIG_HMAC,
verifier=token.verifier)
if not base_url:
base_url = self.base_url
url, headers, body = client.sign(base_url + endpoint,
http_method='POST')
headers.update({'Content-Type': 'application/json'})
return endpoint, headers
def _get_oauth_token(self, consumer, token):
client = oauth1.Client(consumer['key'],
client_secret=consumer['secret'],
resource_owner_key=token.key,
resource_owner_secret=token.secret,
signature_method=oauth1.SIG_HMAC)
endpoint = '/auth/tokens'
url, headers, body = client.sign(self.base_url + endpoint,
http_method='POST')
headers.update({'Content-Type': 'application/json'})
ref = {'auth': {'identity': {'oauth1': {}, 'methods': ['oauth1']}}}
return endpoint, headers, ref
def _authorize_request_token(self, request_id):
if isinstance(request_id, bytes):
request_id = request_id.decode()
return '/OS-OAUTH1/authorize/%s' % (request_id)
def _get_access_token(self):
consumer = self._create_single_consumer()
consumer_id = consumer['id']
consumer_secret = consumer['secret']
consumer = {'key': consumer_id, 'secret': consumer_secret}
url, headers = self._create_request_token(consumer, self.project_id)
content = self.post(
url, headers=headers,
response_content_type='application/x-www-form-urlencoded')
credentials = self._urllib_parse_qs_text_keys(content.result)
request_key = credentials['oauth_token'][0]
request_secret = credentials['oauth_token_secret'][0]
request_token = oauth1.Token(request_key, request_secret)
url = self._authorize_request_token(request_key)
body = {'roles': [{'id': self.role_id}]}
resp = self.put(url, body=body, expected_status=http_client.OK)
verifier = resp.result['token']['oauth_verifier']
request_token.set_verifier(verifier)
url, headers = self._create_access_token(consumer, request_token)
content = self.post(
url, headers=headers,
response_content_type='application/x-www-form-urlencoded')
credentials = self._urllib_parse_qs_text_keys(content.result)
access_key = credentials['oauth_token'][0]
access_secret = credentials['oauth_token_secret'][0]
access_token = oauth1.Token(access_key, access_secret)
url, headers, body = self._get_oauth_token(consumer, access_token)
content = self.post(url, headers=headers, body=body)
return access_key, content.headers['X-Subject-Token']
def test_access_token_ec2_credential(self):
"""Test creating ec2 credential from an oauth access token.
Call ``POST /credentials``.
"""
access_key, token_id = self._get_access_token()
# Create the credential with the access token
blob, ref = unit.new_ec2_credential(user_id=self.user_id,
project_id=self.project_id)
r = self.post('/credentials', body={'credential': ref}, token=token_id)
# We expect the response blob to contain the access_token_id
ret_ref = ref.copy()
ret_blob = blob.copy()
ret_blob['access_token_id'] = access_key.decode('utf-8')
ret_ref['blob'] = json.dumps(ret_blob)
self.assertValidCredentialResponse(r, ref=ret_ref)
# Assert credential id is same as hash of access key id for
# ec2 credentials
access = blob['access'].encode('utf-8')
self.assertEqual(hashlib.sha256(access).hexdigest(),
r.result['credential']['id'])
# Create a role assignment to ensure that it is ignored and only the
# roles in the access token are used
role = unit.new_role_ref(name='reader')
role_id = role['id']
PROVIDERS.role_api.create_role(role_id, role)
PROVIDERS.assignment_api.add_role_to_user_and_project(
self.user_id, self.project_id, role_id)
ret_blob = json.loads(r.result['credential']['blob'])
ec2token = self._test_get_token(
access=ret_blob['access'], secret=ret_blob['secret'])
ec2_roles = [rl['id'] for rl in ec2token['roles']]
self.assertIn(self.role_id, ec2_roles)
self.assertNotIn(role_id, ec2_roles)
class TestCredentialEc2(CredentialBaseTestCase):
"""Test v3 credential compatibility with ec2tokens."""
def _validate_signature(self, access, secret):
"""Test signature validation with the access/secret provided."""
signer = ec2_utils.Ec2Signer(secret)
params = {'SignatureMethod': 'HmacSHA256',
'SignatureVersion': '2',
'AWSAccessKeyId': access}
request = {'host': 'foo',
'verb': 'GET',
'path': '/bar',
'params': params}
signature = signer.generate(request)
# Now make a request to validate the signed dummy request via the
# ec2tokens API. This proves the v3 ec2 credentials actually work.
sig_ref = {'access': access,
'signature': signature,
'host': 'foo',
'verb': 'GET',
'path': '/bar',
'params': params}
r = self.post(
'/ec2tokens',
body={'ec2Credentials': sig_ref},
expected_status=http_client.OK)
self.assertValidTokenResponse(r)
def test_ec2_credential_signature_validate(self):
"""Test signature validation with a v3 ec2 credential."""
blob, ref = unit.new_ec2_credential(user_id=self.user['id'],
@ -502,15 +952,15 @@ class TestCredentialEc2(CredentialBaseTestCase):
cred_blob = json.loads(r.result['credential']['blob'])
self.assertEqual(blob, cred_blob)
self._validate_signature(access=cred_blob['access'],
secret=cred_blob['secret'])
self._test_get_token(access=cred_blob['access'],
secret=cred_blob['secret'])
def test_ec2_credential_signature_validate_legacy(self):
"""Test signature validation with a legacy v3 ec2 credential."""
cred_json, _ = self._create_dict_blob_credential()
cred_blob = json.loads(cred_json)
self._validate_signature(access=cred_blob['access'],
secret=cred_blob['secret'])
self._test_get_token(access=cred_blob['access'],
secret=cred_blob['secret'])
def _get_ec2_cred_uri(self):
return '/users/%s/credentials/OS-EC2' % self.user_id
@ -526,8 +976,8 @@ class TestCredentialEc2(CredentialBaseTestCase):
self.assertEqual(self.user_id, ec2_cred['user_id'])
self.assertEqual(self.project_id, ec2_cred['tenant_id'])
self.assertIsNone(ec2_cred['trust_id'])
self._validate_signature(access=ec2_cred['access'],
secret=ec2_cred['secret'])
self._test_get_token(access=ec2_cred['access'],
secret=ec2_cred['secret'])
uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
self.assertThat(ec2_cred['links']['self'],
matchers.EndsWith(uri))

3
keystone/token/providers/common.py
View File

@ -582,7 +582,8 @@ class BaseProvider(provider_api.ProviderAPIMixin, base.Provider):
auth_context, project_id, domain_id)
access_token = None
if 'oauth1' in method_names:
if 'oauth1' in method_names or (
auth_context and auth_context.get('access_token_id')):
access_token_id = auth_context['access_token_id']
access_token = PROVIDERS.oauth_api.get_access_token(
access_token_id

16
releasenotes/notes/bug-1872733-2377f456a57ad32c.yaml Normal file
View File

@ -0,0 +1,16 @@
---
critical:
- |
[`bug 1872733 <https://bugs.launchpad.net/keystone/+bug/1872733>`_]
Fixed a critical security issue in which an authenticated user could
escalate their privileges by altering a valid EC2 credential.
security:
- |
[`bug 1872733 <https://bugs.launchpad.net/keystone/+bug/1872733>`_]
Fixed a critical security issue in which an authenticated user could
escalate their privileges by altering a valid EC2 credential.
fixes:
- |
[`bug 1872733 <https://bugs.launchpad.net/keystone/+bug/1872733>`_]
Fixed a critical security issue in which an authenticated user could
escalate their privileges by altering a valid EC2 credential.

31
releasenotes/notes/bug-1872735-0989e51d2248ce1e.yaml Normal file
View File

@ -0,0 +1,31 @@
---
critical:
- |
[`bug 1872735 <https://bugs.launchpad.net/keystone/+bug/1872735>`_]
Fixed a security issue in which a trustee or an application credential user
could create an EC2 credential or an application credential that would
permit them to get a token that elevated their role assignments beyond the
subset delegated to them in the trust or application credential. A new
attribute ``app_cred_id`` is now automatically added to the access blob of
an EC2 credential and the role list in the trust or application credential
is respected.
security:
- |
[`bug 1872735 <https://bugs.launchpad.net/keystone/+bug/1872735>`_]
Fixed a security issue in which a trustee or an application credential user
could create an EC2 credential or an application credential that would
permit them to get a token that elevated their role assignments beyond the
subset delegated to them in the trust or application credential. A new
attribute ``app_cred_id`` is now automatically added to the access blob of
an EC2 credential and the role list in the trust or application credential
is respected.
fixes:
- |
[`bug 1872735 <https://bugs.launchpad.net/keystone/+bug/1872735>`_]
Fixed a security issue in which a trustee or an application credential user
could create an EC2 credential or an application credential that would
permit them to get a token that elevated their role assignments beyond the
subset delegated to them in the trust or application credential. A new
attribute ``app_cred_id`` is now automatically added to the access blob of
an EC2 credential and the role list in the trust or application credential
is respected.

19
releasenotes/notes/bug-1872755-2c81d3267b89f124.yaml Normal file
View File

@ -0,0 +1,19 @@
---
security:
- |
[`bug 1872755 <https://bugs.launchpad.net/keystone/+bug/1872755>`_]
Added validation to the EC2 credentials update API to ensure the metadata
labels 'trust_id' and 'app_cred_id' are not altered by the user. These
labels are used by keystone to determine the scope allowed by the
credential, and altering these automatic labels could enable an EC2
credential holder to elevate their access beyond what is permitted by the
application credential or trust that was used to create the EC2 credential.
fixes:
- |
[`bug 1872755 <https://bugs.launchpad.net/keystone/+bug/1872755>`_]
Added validation to the EC2 credentials update API to ensure the metadata
labels 'trust_id' and 'app_cred_id' are not altered by the user. These
labels are used by keystone to determine the scope allowed by the
credential, and altering these automatic labels could enable an EC2
credential holder to elevate their access beyond what is permitted by the
application credential or trust that was used to create the EC2 credential.