54590544fb
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 ine2d83ae9, pep8 fixes ine2d83ae9, test helper in52da4d0e12): keystone/api/credentials.py keystone/tests/unit/test_v3_application_credential.py keystone/tests/unit/test_v3_credential.py Depends-on: https://review.opendev.org/726526 Change-Id: I39d0d705839fbe31ac518ac9a82959e108cb7c1d Closes-bug: #1872733 Closes-bug: #1872755 Closes-bug: #1872735 (cherry picked from commit37e9907a17) (cherry picked from commit2f2736ebb2)
962 lines
40 KiB
Python
962 lines
40 KiB
Python
# Copyright 2013 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.
|
|
|
|
import hashlib
|
|
import json
|
|
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, urllib
|
|
from testtools import matchers
|
|
|
|
from keystone.api import ec2tokens
|
|
from keystone.common import provider_api
|
|
from keystone.common import utils
|
|
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 import test_v3
|
|
|
|
|
|
PROVIDERS = provider_api.ProviderAPIs
|
|
CRED_TYPE_EC2 = ec2tokens.CRED_TYPE_EC2
|
|
|
|
|
|
class CredentialBaseTestCase(test_v3.RestfulTestCase):
|
|
|
|
def setUp(self):
|
|
super(CredentialBaseTestCase, self).setUp()
|
|
self.useFixture(
|
|
ksfixtures.KeyRepository(
|
|
self.config_fixture,
|
|
'credential',
|
|
credential_fernet.MAX_ACTIVE_KEYS
|
|
)
|
|
)
|
|
|
|
def _create_dict_blob_credential(self):
|
|
blob, credential = unit.new_ec2_credential(user_id=self.user['id'],
|
|
project_id=self.project_id)
|
|
|
|
# Store the blob as a dict *not* JSON ref bug #1259584
|
|
# This means we can test the dict->json workaround, added
|
|
# as part of the bugfix for backwards compatibility works.
|
|
credential['blob'] = blob
|
|
credential_id = credential['id']
|
|
|
|
# Create direct via the DB API to avoid validation failure
|
|
PROVIDERS.credential_api.create_credential(credential_id, credential)
|
|
|
|
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."""
|
|
|
|
def setUp(self):
|
|
|
|
super(CredentialTestCase, self).setUp()
|
|
|
|
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_credential_api_delete_credentials_for_project(self):
|
|
PROVIDERS.credential_api.delete_credentials_for_project(
|
|
self.project_id
|
|
)
|
|
# Test that the credential that we created in .setUp no longer exists
|
|
# once we delete all credentials for self.project_id
|
|
self.assertRaises(exception.CredentialNotFound,
|
|
PROVIDERS.credential_api.get_credential,
|
|
credential_id=self.credential['id'])
|
|
|
|
def test_credential_api_delete_credentials_for_user(self):
|
|
PROVIDERS.credential_api.delete_credentials_for_user(self.user_id)
|
|
# Test that the credential that we created in .setUp no longer exists
|
|
# once we delete all credentials for self.user_id
|
|
self.assertRaises(exception.CredentialNotFound,
|
|
PROVIDERS.credential_api.get_credential,
|
|
credential_id=self.credential['id'])
|
|
|
|
def test_list_credentials(self):
|
|
"""Call ``GET /credentials``."""
|
|
r = self.get('/credentials')
|
|
self.assertValidCredentialListResponse(r, ref=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_list_credentials_filtered_by_type(self):
|
|
"""Call ``GET /credentials?type={type}``."""
|
|
PROVIDERS.assignment_api.create_system_grant_for_user(
|
|
self.user_id, self.role_id
|
|
)
|
|
token = self.get_system_scoped_token()
|
|
|
|
# The type ec2 was chosen, instead of a random string,
|
|
# 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=CRED_TYPE_EC2)
|
|
|
|
ec2_resp = PROVIDERS.credential_api.create_credential(
|
|
ec2_credential['id'], ec2_credential)
|
|
|
|
# The type cert was chosen for the same reason as ec2
|
|
r = self.get('/credentials?type=cert', token=token)
|
|
|
|
# Testing the filter for two different types
|
|
self.assertValidCredentialListResponse(r, ref=self.credential)
|
|
for cred in r.result['credentials']:
|
|
self.assertEqual('cert', cred['type'])
|
|
|
|
r_ec2 = self.get('/credentials?type=ec2', token=token)
|
|
self.assertThat(r_ec2.result['credentials'], matchers.HasLength(1))
|
|
cred_ec2 = r_ec2.result['credentials'][0]
|
|
|
|
self.assertValidCredentialListResponse(r_ec2, ref=ec2_resp)
|
|
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):
|
|
"""Call ``GET /credentials?user_id={user_id}&type={type}``."""
|
|
user1_id = uuid.uuid4().hex
|
|
user2_id = uuid.uuid4().hex
|
|
|
|
PROVIDERS.assignment_api.create_system_grant_for_user(
|
|
self.user_id, self.role_id
|
|
)
|
|
token = self.get_system_scoped_token()
|
|
|
|
# Creating credentials for two different users
|
|
credential_user1_ec2 = unit.new_credential_ref(user_id=user1_id,
|
|
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)
|
|
|
|
PROVIDERS.credential_api.create_credential(
|
|
credential_user1_ec2['id'], credential_user1_ec2)
|
|
PROVIDERS.credential_api.create_credential(
|
|
credential_user1_cert['id'], credential_user1_cert)
|
|
PROVIDERS.credential_api.create_credential(
|
|
credential_user2_cert['id'], credential_user2_cert)
|
|
|
|
r = self.get(
|
|
'/credentials?user_id=%s&type=ec2' % user1_id, token=token
|
|
)
|
|
self.assertValidCredentialListResponse(r, ref=credential_user1_ec2)
|
|
self.assertThat(r.result['credentials'], matchers.HasLength(1))
|
|
cred = r.result['credentials'][0]
|
|
self.assertEqual(CRED_TYPE_EC2, cred['type'])
|
|
self.assertEqual(user1_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_update_credential_to_ec2_type(self):
|
|
"""Call ``PATCH /credentials/{credential_id}``."""
|
|
# Create a credential without providing a project_id
|
|
ref = unit.new_credential_ref(user_id=self.user['id'])
|
|
r = self.post(
|
|
'/credentials',
|
|
body={'credential': ref})
|
|
self.assertValidCredentialResponse(r, ref)
|
|
credential_id = r.result.get('credential')['id']
|
|
|
|
# Updating the credential to ec2 requires a project_id
|
|
update_ref = {'type': 'ec2', 'project_id': self.project_id}
|
|
self.patch(
|
|
'/credentials/%(credential_id)s' % {
|
|
'credential_id': credential_id},
|
|
body={'credential': update_ref})
|
|
|
|
def test_update_credential_to_ec2_missing_project_id(self):
|
|
"""Call ``PATCH /credentials/{credential_id}``."""
|
|
# Create a credential without providing a project_id
|
|
ref = unit.new_credential_ref(user_id=self.user['id'])
|
|
r = self.post(
|
|
'/credentials',
|
|
body={'credential': ref})
|
|
self.assertValidCredentialResponse(r, ref)
|
|
credential_id = r.result.get('credential')['id']
|
|
|
|
# Updating such credential to ec2 type without providing a project_id
|
|
# will fail
|
|
update_ref = {'type': 'ec2'}
|
|
self.patch(
|
|
'/credentials/%(credential_id)s' % {
|
|
'credential_id': credential_id},
|
|
body={'credential': update_ref},
|
|
expected_status=http_client.BAD_REQUEST)
|
|
|
|
def test_update_credential_to_ec2_with_previously_set_project_id(self):
|
|
"""Call ``PATCH /credentials/{credential_id}``."""
|
|
# Create a credential providing a project_id
|
|
ref = unit.new_credential_ref(user_id=self.user['id'],
|
|
project_id=self.project_id)
|
|
r = self.post(
|
|
'/credentials',
|
|
body={'credential': ref})
|
|
self.assertValidCredentialResponse(r, ref)
|
|
credential_id = r.result.get('credential')['id']
|
|
|
|
# Since the created credential above already has a project_id, the
|
|
# update request will not fail
|
|
update_ref = {'type': 'ec2'}
|
|
self.patch(
|
|
'/credentials/%(credential_id)s' % {
|
|
'credential_id': credential_id},
|
|
body={'credential': update_ref})
|
|
|
|
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})
|
|
|
|
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(
|
|
'/credentials/%(credential_id)s' % {
|
|
'credential_id': self.credential['id']})
|
|
|
|
def test_delete_credential_retries_on_deadlock(self):
|
|
patcher = mock.patch('sqlalchemy.orm.query.Query.delete',
|
|
autospec=True)
|
|
|
|
class FakeDeadlock(object):
|
|
def __init__(self, mock_patcher):
|
|
self.deadlock_count = 2
|
|
self.mock_patcher = mock_patcher
|
|
self.patched = True
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
if self.deadlock_count > 1:
|
|
self.deadlock_count -= 1
|
|
else:
|
|
self.mock_patcher.stop()
|
|
self.patched = False
|
|
raise oslo_db_exception.DBDeadlock
|
|
|
|
sql_delete_mock = patcher.start()
|
|
side_effect = FakeDeadlock(patcher)
|
|
sql_delete_mock.side_effect = side_effect
|
|
|
|
try:
|
|
PROVIDERS.credential_api.delete_credentials_for_user(
|
|
user_id=self.user['id'])
|
|
finally:
|
|
if side_effect.patched:
|
|
patcher.stop()
|
|
|
|
# initial attempt + 1 retry
|
|
self.assertEqual(sql_delete_mock.call_count, 2)
|
|
|
|
def test_create_ec2_credential(self):
|
|
"""Call ``POST /credentials`` for creating ec2 credential."""
|
|
blob, ref = unit.new_ec2_credential(user_id=self.user['id'],
|
|
project_id=self.project_id)
|
|
r = self.post('/credentials', body={'credential': ref})
|
|
self.assertValidCredentialResponse(r, 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 second ec2 credential with the same access key id and check
|
|
# for conflict.
|
|
self.post(
|
|
'/credentials',
|
|
body={'credential': ref}, expected_status=http_client.CONFLICT)
|
|
|
|
def test_get_ec2_dict_blob(self):
|
|
"""Ensure non-JSON blob data is correctly converted."""
|
|
expected_blob, credential_id = self._create_dict_blob_credential()
|
|
|
|
r = self.get(
|
|
'/credentials/%(credential_id)s' % {
|
|
'credential_id': credential_id})
|
|
|
|
# use json.loads to transform the blobs back into Python dictionaries
|
|
# to avoid problems with the keys being in different orders.
|
|
self.assertEqual(json.loads(expected_blob),
|
|
json.loads(r.result['credential']['blob']))
|
|
|
|
def test_list_ec2_dict_blob(self):
|
|
"""Ensure non-JSON blob data is correctly converted."""
|
|
expected_blob, credential_id = self._create_dict_blob_credential()
|
|
|
|
list_r = self.get('/credentials')
|
|
list_creds = list_r.result['credentials']
|
|
list_ids = [r['id'] for r in list_creds]
|
|
self.assertIn(credential_id, list_ids)
|
|
# use json.loads to transform the blobs back into Python dictionaries
|
|
# to avoid problems with the keys being in different orders.
|
|
for r in list_creds:
|
|
if r['id'] == credential_id:
|
|
self.assertEqual(json.loads(expected_blob),
|
|
json.loads(r['blob']))
|
|
|
|
def test_create_non_ec2_credential(self):
|
|
"""Test creating non-ec2 credential.
|
|
|
|
Call ``POST /credentials``.
|
|
"""
|
|
blob, ref = unit.new_cert_credential(user_id=self.user['id'])
|
|
|
|
r = self.post('/credentials', body={'credential': ref})
|
|
self.assertValidCredentialResponse(r, ref)
|
|
# Assert credential id is not same as hash of access key id for
|
|
# non-ec2 credentials
|
|
access = blob['access'].encode('utf-8')
|
|
self.assertNotEqual(hashlib.sha256(access).hexdigest(),
|
|
r.result['credential']['id'])
|
|
|
|
def test_create_ec2_credential_with_missing_project_id(self):
|
|
"""Test Creating ec2 credential with missing project_id.
|
|
|
|
Call ``POST /credentials``.
|
|
"""
|
|
_, ref = unit.new_ec2_credential(user_id=self.user['id'],
|
|
project_id=None)
|
|
# Assert bad request status when missing project_id
|
|
self.post(
|
|
'/credentials',
|
|
body={'credential': ref}, expected_status=http_client.BAD_REQUEST)
|
|
|
|
def test_create_ec2_credential_with_invalid_blob(self):
|
|
"""Test creating ec2 credential with invalid blob.
|
|
|
|
Call ``POST /credentials``.
|
|
"""
|
|
ref = unit.new_credential_ref(user_id=self.user['id'],
|
|
project_id=self.project_id,
|
|
blob='{"abc":"def"d}',
|
|
type=CRED_TYPE_EC2)
|
|
# Assert bad request status when request contains invalid blob
|
|
response = self.post(
|
|
'/credentials',
|
|
body={'credential': ref}, expected_status=http_client.BAD_REQUEST)
|
|
self.assertValidErrorResponse(response)
|
|
|
|
def test_create_credential_with_admin_token(self):
|
|
# Make sure we can create credential with the static admin token
|
|
ref = unit.new_credential_ref(user_id=self.user['id'])
|
|
r = self.post(
|
|
'/credentials',
|
|
body={'credential': ref},
|
|
token=self.get_admin_token())
|
|
self.assertValidCredentialResponse(r, ref)
|
|
|
|
|
|
class TestCredentialTrustScoped(CredentialBaseTestCase):
|
|
"""Test credential with trust scoped token."""
|
|
|
|
def setUp(self):
|
|
super(TestCredentialTrustScoped, self).setUp()
|
|
|
|
self.trustee_user = unit.new_user_ref(domain_id=self.domain_id)
|
|
password = self.trustee_user['password']
|
|
self.trustee_user = PROVIDERS.identity_api.create_user(
|
|
self.trustee_user
|
|
)
|
|
self.trustee_user['password'] = password
|
|
self.trustee_user_id = self.trustee_user['id']
|
|
self.useFixture(
|
|
ksfixtures.KeyRepository(
|
|
self.config_fixture,
|
|
'credential',
|
|
credential_fernet.MAX_ACTIVE_KEYS
|
|
)
|
|
)
|
|
|
|
def config_overrides(self):
|
|
super(TestCredentialTrustScoped, self).config_overrides()
|
|
self.config_fixture.config(group='trust')
|
|
|
|
def test_trust_scoped_ec2_credential(self):
|
|
"""Test creating trust scoped ec2 credential.
|
|
|
|
Call ``POST /credentials``.
|
|
"""
|
|
# Create the trust
|
|
ref = unit.new_trust_ref(
|
|
trustor_user_id=self.user_id,
|
|
trustee_user_id=self.trustee_user_id,
|
|
project_id=self.project_id,
|
|
impersonation=True,
|
|
expires=dict(minutes=1),
|
|
role_ids=[self.role_id])
|
|
del ref['id']
|
|
r = self.post('/OS-TRUST/trusts', body={'trust': ref})
|
|
trust = self.assertValidTrustResponse(r)
|
|
|
|
# Get a trust scoped token
|
|
auth_data = self.build_authentication_request(
|
|
user_id=self.trustee_user['id'],
|
|
password=self.trustee_user['password'],
|
|
trust_id=trust['id'])
|
|
r = self.v3_create_token(auth_data)
|
|
self.assertValidProjectScopedTokenResponse(r, self.user)
|
|
trust_id = r.result['token']['OS-TRUST:trust']['id']
|
|
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,
|
|
project_id=self.project_id)
|
|
r = self.post('/credentials', body={'credential': ref}, token=token_id)
|
|
|
|
# We expect the response blob to contain the trust_id
|
|
ret_ref = ref.copy()
|
|
ret_blob = blob.copy()
|
|
ret_blob['trust_id'] = trust_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
|
|
# 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 = [role['id'] for role 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 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 = unit.new_application_credential_ref(roles=[{'id': self.role_id}])
|
|
del ref['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 = [role['id'] for role 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 = [role['id'] for role 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 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'],
|
|
project_id=self.project_id)
|
|
r = self.post('/credentials', body={'credential': ref})
|
|
self.assertValidCredentialResponse(r, ref)
|
|
# Assert credential id is same as hash of access key id
|
|
access = blob['access'].encode('utf-8')
|
|
self.assertEqual(hashlib.sha256(access).hexdigest(),
|
|
r.result['credential']['id'])
|
|
|
|
cred_blob = json.loads(r.result['credential']['blob'])
|
|
self.assertEqual(blob, cred_blob)
|
|
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._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
|
|
|
|
def _get_ec2_cred(self):
|
|
uri = self._get_ec2_cred_uri()
|
|
r = self.post(uri, body={'tenant_id': self.project_id})
|
|
return r.result['credential']
|
|
|
|
def test_ec2_create_credential(self):
|
|
"""Test ec2 credential creation."""
|
|
ec2_cred = self._get_ec2_cred()
|
|
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._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))
|
|
|
|
def test_ec2_get_credential(self):
|
|
ec2_cred = self._get_ec2_cred()
|
|
uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
|
|
r = self.get(uri)
|
|
self.assertDictEqual(ec2_cred, r.result['credential'])
|
|
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
|
|
PROVIDERS.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()
|
|
uri = self._get_ec2_cred_uri()
|
|
r = self.get(uri)
|
|
cred_list = r.result['credentials']
|
|
self.assertEqual(1, len(cred_list))
|
|
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
|
|
PROVIDERS.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 = (
|
|
PROVIDERS.credential_api
|
|
.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,
|
|
PROVIDERS.credential_api.get_credential,
|
|
cred_from_credential_api[0]['id'])
|