openstack
/
keystone
Code Issues Proposed changes
keystone/keystone/tests/unit/test_v3_credential.py

479 lines
20 KiB
Python
Raw Blame History

# 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
from oslo_config import cfg
from six.moves import http_client
from testtools import matchers
from keystone.common import utils
from keystone.contrib.ec2 import controllers
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import test_v3
CONF = cfg.CONF
CRED_TYPE_EC2 = controllers.CRED_TYPE_EC2
class CredentialBaseTestCase(test_v3.RestfulTestCase):
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
self.credential_api.create_credential(credential_id, credential)
return json.dumps(blob), credential_id
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)
self.credential_api.create_credential(
self.credential['id'],
self.credential)
def test_credential_api_delete_credentials_for_project(self):
self.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,
self.credential_api.get_credential,
credential_id=self.credential['id'])
def test_credential_api_delete_credentials_for_user(self):
self.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,
self.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)
self.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}``."""
# 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 = self.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')
# 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')
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
# 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)
self.credential_api.create_credential(
credential_user1_ec2['id'], credential_user1_ec2)
self.credential_api.create_credential(
credential_user1_cert['id'], credential_user1_cert)
self.credential_api.create_credential(
credential_user2_cert['id'], credential_user2_cert)
r = self.get('/credentials?user_id=%s&type=ec2' % user1_id)
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_delete_credential(self):
"""Call ``DELETE /credentials/{credential_id}``."""
self.delete(
'/credentials/%(credential_id)s' % {
'credential_id': self.credential['id']})
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(test_v3.RestfulTestCase):
"""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 = self.identity_api.create_user(self.trustee_user)
self.trustee_user['password'] = password
self.trustee_user_id = self.trustee_user['id']
def config_overrides(self):
super(TestCredentialTrustScoped, self).config_overrides()
self.config_fixture.config(group='trust', enabled=True)
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 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 TestCredentialEc2(CredentialBaseTestCase):
"""Test v3 credential compatibility with ec2tokens."""
def setUp(self):
super(TestCredentialEc2, self).setUp()
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'],
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._validate_signature(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'])
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._validate_signature(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
self.credential_api.create_credential(cred_id, non_ec2_cred)
uri = '/'.join([self._get_ec2_cred_uri(), access_key])
# if access_key is not found, ec2 controller raises Unauthorized
# exception
self.get(uri, expected_status=http_client.UNAUTHORIZED)
def test_ec2_list_credentials(self):
"""Test ec2 credential listing."""
self._get_ec2_cred()
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
self.credential_api.create_credential(non_ec2_cred['id'],
non_ec2_cred)
r = self.get(uri)
cred_list_2 = r.result['credentials']
# still one element because non-EC2 credentials are not returned.
self.assertEqual(1, len(cred_list_2))
self.assertEqual(cred_list[0], cred_list_2[0])
def test_ec2_delete_credential(self):
"""Test ec2 credential deletion."""
ec2_cred = self._get_ec2_cred()
uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
cred_from_credential_api = (
self.credential_api
.list_credentials_for_user(self.user_id, type=CRED_TYPE_EC2))
self.assertEqual(1, len(cred_from_credential_api))
self.delete(uri)
self.assertRaises(exception.CredentialNotFound,
self.credential_api.get_credential,
cred_from_credential_api[0]['id'])
View Git Blame Copy Permalink