964 lines
40 KiB
Python
964 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 = {
|
|
'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 = [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'])
|