From 5e02b8b6e49244659c88f9f9b4cb6789d820e559 Mon Sep 17 00:00:00 2001 From: andreaf Date: Wed, 18 Mar 2015 19:03:34 +0000 Subject: [PATCH] Migrated auth.py and related unit tests from tempest Migrated auth.py, test_auth.py, fake_credentials.py, fake_identity.py, fake_http.py from tempest, as well as the EndpointNotFound exception. This migrates the above files from tempest. This latest Change-Ids of each file at the time of this migration of each fare the following: * auth.py : If5a889be5826d60bf61dbb84661a5896cb094875 * tests/test_auth.py : Iedcd2fa7713409743b1b4b9f460481608796d491 * tests/fake_credentials.py : If5a889be5826d60bf61dbb84661a5896cb094875 * tests/fake_identity.py : Ie405ba6a71d29258e99d2f0b68afb013a9619e9b * tests/fake_http.py : I8f14cd2ca6afc38d3fe8ee758272071111022896 to see the commit history for these files refer to the above sha1s in the tempest repository Change-Id: I073d728a2cbf2f6ddde0a8629cc53eccbe3d6d8f --- tempest_lib/auth.py | 648 ++++++++++++++++++++++++++ tempest_lib/exceptions.py | 6 +- tempest_lib/tests/fake_credentials.py | 59 +++ tempest_lib/tests/fake_http.py | 2 +- tempest_lib/tests/fake_identity.py | 164 +++++++ tempest_lib/tests/test_auth.py | 407 ++++++++++++++++ 6 files changed, 1284 insertions(+), 2 deletions(-) create mode 100644 tempest_lib/auth.py create mode 100644 tempest_lib/tests/fake_credentials.py create mode 100644 tempest_lib/tests/fake_identity.py create mode 100644 tempest_lib/tests/test_auth.py diff --git a/tempest_lib/auth.py b/tempest_lib/auth.py new file mode 100644 index 0000000..dca496d --- /dev/null +++ b/tempest_lib/auth.py @@ -0,0 +1,648 @@ +# Copyright 2014 Hewlett-Packard Development Company, L.P. +# All Rights Reserved. +# +# 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 abc +import copy +import datetime +import re +try: + import urllib.parse as urlparse +except ImportError: + import urlparse + +from oslo_log import log as logging +import six + +from tempest_lib import exceptions +from tempest_lib.services.identity.v2 import token_client as json_v2id +from tempest_lib.services.identity.v3 import token_client as json_v3id + + +LOG = logging.getLogger(__name__) + + +@six.add_metaclass(abc.ABCMeta) +class AuthProvider(object): + """Provide authentication""" + + def __init__(self, credentials): + """Auth provider __init__ + + :param credentials: credentials for authentication + """ + if self.check_credentials(credentials): + self.credentials = credentials + else: + raise TypeError("Invalid credentials") + self.cache = None + self.alt_auth_data = None + self.alt_part = None + + def __str__(self): + return "Creds :{creds}, cached auth data: {cache}".format( + creds=self.credentials, cache=self.cache) + + @abc.abstractmethod + def _decorate_request(self, filters, method, url, headers=None, body=None, + auth_data=None): + """Decorate request with authentication data""" + return + + @abc.abstractmethod + def _get_auth(self): + return + + @abc.abstractmethod + def _fill_credentials(self, auth_data_body): + return + + def fill_credentials(self): + """Fill credentials object with data from auth""" + auth_data = self.get_auth() + self._fill_credentials(auth_data[1]) + return self.credentials + + @classmethod + def check_credentials(cls, credentials): + """Verify credentials are valid.""" + return isinstance(credentials, Credentials) and credentials.is_valid() + + @property + def auth_data(self): + return self.get_auth() + + @auth_data.deleter + def auth_data(self): + self.clear_auth() + + def get_auth(self): + """Returns auth from cache if available, else auth first""" + if self.cache is None or self.is_expired(self.cache): + self.set_auth() + return self.cache + + def set_auth(self): + """Forces setting auth. + + Forces setting auth, ignores cache if it exists. + Refills credentials + """ + self.cache = self._get_auth() + self._fill_credentials(self.cache[1]) + + def clear_auth(self): + """Clear access cache + + Can be called to clear the access cache so that next request + will fetch a new token and base_url. + """ + self.cache = None + self.credentials.reset() + + @abc.abstractmethod + def is_expired(self, auth_data): + return + + def auth_request(self, method, url, headers=None, body=None, filters=None): + """Obtains auth data and decorates a request with that. + + :param method: HTTP method of the request + :param url: relative URL of the request (path) + :param headers: HTTP headers of the request + :param body: HTTP body in case of POST / PUT + :param filters: select a base URL out of the catalog + :returns a Tuple (url, headers, body) + """ + orig_req = dict(url=url, headers=headers, body=body) + + auth_url, auth_headers, auth_body = self._decorate_request( + filters, method, url, headers, body) + auth_req = dict(url=auth_url, headers=auth_headers, body=auth_body) + + # Overwrite part if the request if it has been requested + if self.alt_part is not None: + if self.alt_auth_data is not None: + alt_url, alt_headers, alt_body = self._decorate_request( + filters, method, url, headers, body, + auth_data=self.alt_auth_data) + alt_auth_req = dict(url=alt_url, headers=alt_headers, + body=alt_body) + auth_req[self.alt_part] = alt_auth_req[self.alt_part] + + else: + # If alt auth data is None, skip auth in the requested part + auth_req[self.alt_part] = orig_req[self.alt_part] + + # Next auth request will be normal, unless otherwise requested + self.reset_alt_auth_data() + + return auth_req['url'], auth_req['headers'], auth_req['body'] + + def reset_alt_auth_data(self): + """Configure auth provider to provide valid authentication data""" + self.alt_part = None + self.alt_auth_data = None + + def set_alt_auth_data(self, request_part, auth_data): + """Alternate auth data on next request + + Configure auth provider to provide alt authentication data + on a part of the *next* auth_request. If credentials are None, + set invalid data. + :param request_part: request part to contain invalid auth: url, + headers, body + :param auth_data: alternative auth_data from which to get the + invalid data to be injected + """ + self.alt_part = request_part + self.alt_auth_data = auth_data + + @abc.abstractmethod + def base_url(self, filters, auth_data=None): + """Extracts the base_url based on provided filters""" + return + + +class KeystoneAuthProvider(AuthProvider): + + token_expiry_threshold = datetime.timedelta(seconds=60) + + def __init__(self, credentials, auth_url, + disable_ssl_certificate_validation=None, + ca_certs=None, trace_requests=None): + super(KeystoneAuthProvider, self).__init__(credentials) + self.dsvm = disable_ssl_certificate_validation + self.ca_certs = ca_certs + self.trace_requests = trace_requests + self.auth_client = self._auth_client(auth_url) + + def _decorate_request(self, filters, method, url, headers=None, body=None, + auth_data=None): + if auth_data is None: + auth_data = self.auth_data + token, _ = auth_data + base_url = self.base_url(filters=filters, auth_data=auth_data) + # build authenticated request + # returns new request, it does not touch the original values + _headers = copy.deepcopy(headers) if headers is not None else {} + _headers['X-Auth-Token'] = str(token) + if url is None or url == "": + _url = base_url + else: + # Join base URL and url, and remove multiple contiguous slashes + _url = "/".join([base_url, url]) + parts = [x for x in urlparse.urlparse(_url)] + parts[2] = re.sub("/{2,}", "/", parts[2]) + _url = urlparse.urlunparse(parts) + # no change to method or body + return str(_url), _headers, body + + @abc.abstractmethod + def _auth_client(self): + return + + @abc.abstractmethod + def _auth_params(self): + return + + def _get_auth(self): + # Bypasses the cache + auth_func = getattr(self.auth_client, 'get_token') + auth_params = self._auth_params() + + # returns token, auth_data + token, auth_data = auth_func(**auth_params) + return token, auth_data + + def get_token(self): + return self.auth_data[0] + + +class KeystoneV2AuthProvider(KeystoneAuthProvider): + + EXPIRY_DATE_FORMAT = '%Y-%m-%dT%H:%M:%SZ' + + def _auth_client(self, auth_url): + return json_v2id.TokenClientJSON( + auth_url, disable_ssl_certificate_validation=self.dsvm, + ca_certs=self.ca_certs, trace_requests=self.trace_requests) + + def _auth_params(self): + return dict( + user=self.credentials.username, + password=self.credentials.password, + tenant=self.credentials.tenant_name, + auth_data=True) + + def _fill_credentials(self, auth_data_body): + tenant = auth_data_body['token']['tenant'] + user = auth_data_body['user'] + if self.credentials.tenant_name is None: + self.credentials.tenant_name = tenant['name'] + if self.credentials.tenant_id is None: + self.credentials.tenant_id = tenant['id'] + if self.credentials.username is None: + self.credentials.username = user['name'] + if self.credentials.user_id is None: + self.credentials.user_id = user['id'] + + def base_url(self, filters, auth_data=None): + """Base URL from catalog + + Filters can be: + - service: compute, image, etc + - region: the service region + - endpoint_type: adminURL, publicURL, internalURL + - api_version: replace catalog version with this + - skip_path: take just the base URL + """ + if auth_data is None: + auth_data = self.auth_data + token, _auth_data = auth_data + service = filters.get('service') + region = filters.get('region') + endpoint_type = filters.get('endpoint_type', 'publicURL') + + if service is None: + raise exceptions.EndpointNotFound("No service provided") + + _base_url = None + for ep in _auth_data['serviceCatalog']: + if ep["type"] == service: + for _ep in ep['endpoints']: + if region is not None and _ep['region'] == region: + _base_url = _ep.get(endpoint_type) + if not _base_url: + # No region matching, use the first + _base_url = ep['endpoints'][0].get(endpoint_type) + break + if _base_url is None: + raise exceptions.EndpointNotFound(service) + + parts = urlparse.urlparse(_base_url) + if filters.get('api_version', None) is not None: + path = "/" + filters['api_version'] + noversion_path = "/".join(parts.path.split("/")[2:]) + if noversion_path != "": + path += "/" + noversion_path + _base_url = _base_url.replace(parts.path, path) + if filters.get('skip_path', None) is not None and parts.path != '': + _base_url = _base_url.replace(parts.path, "/") + + return _base_url + + def is_expired(self, auth_data): + _, access = auth_data + expiry = datetime.datetime.strptime(access['token']['expires'], + self.EXPIRY_DATE_FORMAT) + return (expiry - self.token_expiry_threshold <= + datetime.datetime.utcnow()) + + +class KeystoneV3AuthProvider(KeystoneAuthProvider): + + EXPIRY_DATE_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ' + + def _auth_client(self, auth_url): + return json_v3id.V3TokenClientJSON( + auth_url, disable_ssl_certificate_validation=self.dsvm, + ca_certs=self.ca_certs, trace_requests=self.trace_requests) + + def _auth_params(self): + return dict( + user_id=self.credentials.user_id, + username=self.credentials.username, + password=self.credentials.password, + project_id=self.credentials.project_id, + project_name=self.credentials.project_name, + user_domain_id=self.credentials.user_domain_id, + user_domain_name=self.credentials.user_domain_name, + project_domain_id=self.credentials.project_domain_id, + project_domain_name=self.credentials.project_domain_name, + domain_id=self.credentials.domain_id, + domain_name=self.credentials.domain_name, + auth_data=True) + + def _fill_credentials(self, auth_data_body): + # project or domain, depending on the scope + project = auth_data_body.get('project', None) + domain = auth_data_body.get('domain', None) + # user is always there + user = auth_data_body['user'] + # Set project fields + if project is not None: + if self.credentials.project_name is None: + self.credentials.project_name = project['name'] + if self.credentials.project_id is None: + self.credentials.project_id = project['id'] + if self.credentials.project_domain_id is None: + self.credentials.project_domain_id = project['domain']['id'] + if self.credentials.project_domain_name is None: + self.credentials.project_domain_name = ( + project['domain']['name']) + # Set domain fields + if domain is not None: + if self.credentials.domain_id is None: + self.credentials.domain_id = domain['id'] + if self.credentials.domain_name is None: + self.credentials.domain_name = domain['name'] + # Set user fields + if self.credentials.username is None: + self.credentials.username = user['name'] + if self.credentials.user_id is None: + self.credentials.user_id = user['id'] + if self.credentials.user_domain_id is None: + self.credentials.user_domain_id = user['domain']['id'] + if self.credentials.user_domain_name is None: + self.credentials.user_domain_name = user['domain']['name'] + + def base_url(self, filters, auth_data=None): + """Base URL from catalog + + Filters can be: + - service: compute, image, etc + - region: the service region + - endpoint_type: adminURL, publicURL, internalURL + - api_version: replace catalog version with this + - skip_path: take just the base URL + """ + if auth_data is None: + auth_data = self.auth_data + token, _auth_data = auth_data + service = filters.get('service') + region = filters.get('region') + endpoint_type = filters.get('endpoint_type', 'public') + + if service is None: + raise exceptions.EndpointNotFound("No service provided") + + if 'URL' in endpoint_type: + endpoint_type = endpoint_type.replace('URL', '') + _base_url = None + catalog = _auth_data['catalog'] + # Select entries with matching service type + service_catalog = [ep for ep in catalog if ep['type'] == service] + if len(service_catalog) > 0: + service_catalog = service_catalog[0]['endpoints'] + else: + # No matching service + raise exceptions.EndpointNotFound(service) + # Filter by endpoint type (interface) + filtered_catalog = [ep for ep in service_catalog if + ep['interface'] == endpoint_type] + if len(filtered_catalog) == 0: + # No matching type, keep all and try matching by region at least + filtered_catalog = service_catalog + # Filter by region + filtered_catalog = [ep for ep in filtered_catalog if + ep['region'] == region] + if len(filtered_catalog) == 0: + # No matching region, take the first endpoint + filtered_catalog = [service_catalog[0]] + # There should be only one match. If not take the first. + _base_url = filtered_catalog[0].get('url', None) + if _base_url is None: + raise exceptions.EndpointNotFound(service) + + parts = urlparse.urlparse(_base_url) + if filters.get('api_version', None) is not None: + path = "/" + filters['api_version'] + noversion_path = "/".join(parts.path.split("/")[2:]) + if noversion_path != "": + path += "/" + noversion_path + _base_url = _base_url.replace(parts.path, path) + if filters.get('skip_path', None) is not None: + _base_url = _base_url.replace(parts.path, "/") + + return _base_url + + def is_expired(self, auth_data): + _, access = auth_data + expiry = datetime.datetime.strptime(access['expires_at'], + self.EXPIRY_DATE_FORMAT) + return (expiry - self.token_expiry_threshold <= + datetime.datetime.utcnow()) + + +def is_identity_version_supported(identity_version): + return identity_version in IDENTITY_VERSION + + +def get_credentials(auth_url, fill_in=True, identity_version='v2', + disable_ssl_certificate_validation=None, ca_certs=None, + trace_requests=None, **kwargs): + """Builds a credentials object based on the configured auth_version + + :param auth_url (string): Full URI of the OpenStack Identity API(Keystone) + which is used to fetch the token from Identity service. + :param fill_in (boolean): obtain a token and fill in all credential + details provided by the identity service. When fill_in is not + specified, credentials are not validated. Validation can be invoked + by invoking ``is_valid()`` + :param identity_version (string): identity API version is used to + select the matching auth provider and credentials class + :param disable_ssl_certificate_validation: whether to enforce SSL + certificate validation in SSL API requests to the auth system + :param ca_certs: CA certificate bundle for validation of certificates + in SSL API requests to the auth system + :param trace_requests: trace in log API requests to the auth system + :param kwargs (dict): Dict of credential key/value pairs + + Examples: + + Returns credentials from the provided parameters: + >>> get_credentials(username='foo', password='bar') + + Returns credentials including IDs: + >>> get_credentials(username='foo', password='bar', fill_in=True) + """ + if not is_identity_version_supported(identity_version): + raise exceptions.InvalidIdentityVersion( + identity_version=identity_version) + + credential_class, auth_provider_class = IDENTITY_VERSION.get( + identity_version) + + creds = credential_class(**kwargs) + # Fill in the credentials fields that were not specified + if fill_in: + dsvm = disable_ssl_certificate_validation + auth_provider = auth_provider_class( + creds, auth_url, disable_ssl_certificate_validation=dsvm, + ca_certs=ca_certs, trace_requests=trace_requests) + creds = auth_provider.fill_credentials() + return creds + + +class Credentials(object): + """Set of credentials for accessing OpenStack services + + ATTRIBUTES: list of valid class attributes representing credentials. + """ + + ATTRIBUTES = [] + + def __init__(self, **kwargs): + """Enforce the available attributes at init time (only). + + Additional attributes can still be set afterwards if tests need + to do so. + """ + self._initial = kwargs + self._apply_credentials(kwargs) + + def _apply_credentials(self, attr): + for key in attr.keys(): + if key in self.ATTRIBUTES: + setattr(self, key, attr[key]) + else: + raise exceptions.InvalidCredentials + + def __str__(self): + """Represent only attributes included in self.ATTRIBUTES""" + _repr = dict((k, getattr(self, k)) for k in self.ATTRIBUTES) + return str(_repr) + + def __eq__(self, other): + """Credentials are equal if attributes in self.ATTRIBUTES are equal""" + return str(self) == str(other) + + def __getattr__(self, key): + # If an attribute is set, __getattr__ is not invoked + # If an attribute is not set, and it is a known one, return None + if key in self.ATTRIBUTES: + return None + else: + raise AttributeError + + def __delitem__(self, key): + # For backwards compatibility, support dict behaviour + if key in self.ATTRIBUTES: + delattr(self, key) + else: + raise AttributeError + + def get(self, item, default): + # In this patch act as dict for backward compatibility + try: + return getattr(self, item) + except AttributeError: + return default + + def get_init_attributes(self): + return self._initial.keys() + + def is_valid(self): + raise NotImplementedError + + def reset(self): + # First delete all known attributes + for key in self.ATTRIBUTES: + if getattr(self, key) is not None: + delattr(self, key) + # Then re-apply initial setup + self._apply_credentials(self._initial) + + +class KeystoneV2Credentials(Credentials): + + ATTRIBUTES = ['username', 'password', 'tenant_name', 'user_id', + 'tenant_id'] + + def is_valid(self): + """Check of credentials (no API call) + + Minimum set of valid credentials, are username and password. + Tenant is optional. + """ + return None not in (self.username, self.password) + + +class KeystoneV3Credentials(Credentials): + """Credentials suitable for the Keystone Identity V3 API""" + + ATTRIBUTES = ['domain_id', 'domain_name', 'password', 'username', + 'project_domain_id', 'project_domain_name', 'project_id', + 'project_name', 'tenant_id', 'tenant_name', 'user_domain_id', + 'user_domain_name', 'user_id'] + + def __setattr__(self, key, value): + parent = super(KeystoneV3Credentials, self) + # for tenant_* set both project and tenant + if key == 'tenant_id': + parent.__setattr__('project_id', value) + elif key == 'tenant_name': + parent.__setattr__('project_name', value) + # for project_* set both project and tenant + if key == 'project_id': + parent.__setattr__('tenant_id', value) + elif key == 'project_name': + parent.__setattr__('tenant_name', value) + # for *_domain_* set both user and project if not set yet + if key == 'user_domain_id': + if self.project_domain_id is None: + parent.__setattr__('project_domain_id', value) + if key == 'project_domain_id': + if self.user_domain_id is None: + parent.__setattr__('user_domain_id', value) + if key == 'user_domain_name': + if self.project_domain_name is None: + parent.__setattr__('project_domain_name', value) + if key == 'project_domain_name': + if self.user_domain_name is None: + parent.__setattr__('user_domain_name', value) + # support domain_name coming from config + if key == 'domain_name': + parent.__setattr__('user_domain_name', value) + parent.__setattr__('project_domain_name', value) + # finally trigger default behaviour for all attributes + parent.__setattr__(key, value) + + def is_valid(self): + """Check of credentials (no API call) + + Valid combinations of v3 credentials (excluding token, scope) + - User id, password (optional domain) + - User name, password and its domain id/name + For the scope, valid combinations are: + - None + - Project id (optional domain) + - Project name and its domain id/name + - Domain id + - Domain name + """ + valid_user_domain = any( + [self.user_domain_id is not None, + self.user_domain_name is not None]) + valid_project_domain = any( + [self.project_domain_id is not None, + self.project_domain_name is not None]) + valid_user = any( + [self.user_id is not None, + self.username is not None and valid_user_domain]) + valid_project_scope = any( + [self.project_name is None and self.project_id is None, + self.project_id is not None, + self.project_name is not None and valid_project_domain]) + valid_domain_scope = any( + [self.domain_id is None and self.domain_name is None, + self.domain_id or self.domain_name]) + return all([self.password is not None, + valid_user, + valid_project_scope and valid_domain_scope]) + + +IDENTITY_VERSION = {'v2': (KeystoneV2Credentials, KeystoneV2AuthProvider), + 'v3': (KeystoneV3Credentials, KeystoneV3AuthProvider)} diff --git a/tempest_lib/exceptions.py b/tempest_lib/exceptions.py index b6672a9..2ceef3e 100644 --- a/tempest_lib/exceptions.py +++ b/tempest_lib/exceptions.py @@ -146,4 +146,8 @@ class CommandFailed(Exception): class IdentityError(TempestException): - message = "Got identity error" \ No newline at end of file + message = "Got identity error" + + +class EndpointNotFound(TempestException): + message = "Endpoint not found" \ No newline at end of file diff --git a/tempest_lib/tests/fake_credentials.py b/tempest_lib/tests/fake_credentials.py new file mode 100644 index 0000000..32105a1 --- /dev/null +++ b/tempest_lib/tests/fake_credentials.py @@ -0,0 +1,59 @@ +# Copyright 2014 Hewlett-Packard Development Company, L.P. +# All Rights Reserved. +# +# 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. + +from tempest_lib import auth + + +class FakeCredentials(auth.Credentials): + + def is_valid(self): + return True + + +class FakeKeystoneV2Credentials(auth.KeystoneV2Credentials): + + def __init__(self): + creds = dict( + username='fake_username', + password='fake_password', + tenant_name='fake_tenant_name' + ) + super(FakeKeystoneV2Credentials, self).__init__(**creds) + + +class FakeKeystoneV3Credentials(auth.KeystoneV3Credentials): + """Fake credentials suitable for the Keystone Identity V3 API""" + + def __init__(self): + creds = dict( + username='fake_username', + password='fake_password', + user_domain_name='fake_domain_name', + project_name='fake_tenant_name', + project_domain_name='fake_domain_name' + ) + super(FakeKeystoneV3Credentials, self).__init__(**creds) + + +class FakeKeystoneV3DomainCredentials(auth.KeystoneV3Credentials): + """Fake credentials for the Keystone Identity V3 API, with no scope""" + + def __init__(self): + creds = dict( + username='fake_username', + password='fake_password', + user_domain_name='fake_domain_name' + ) + super(FakeKeystoneV3DomainCredentials, self).__init__(**creds) diff --git a/tempest_lib/tests/fake_http.py b/tempest_lib/tests/fake_http.py index e14b25c..002f16b 100644 --- a/tempest_lib/tests/fake_http.py +++ b/tempest_lib/tests/fake_http.py @@ -50,7 +50,7 @@ class fake_httplib2(object): class fake_httplib(object): def __init__(self, headers, body=None, version=1.0, status=200, reason="Ok"): - """fake_httplib for tests + """Fake httplib implementation :param headers: dict representing HTTP response headers :param body: file-like object diff --git a/tempest_lib/tests/fake_identity.py b/tempest_lib/tests/fake_identity.py new file mode 100644 index 0000000..ad78f85 --- /dev/null +++ b/tempest_lib/tests/fake_identity.py @@ -0,0 +1,164 @@ +# Copyright 2014 IBM Corp. +# All Rights Reserved. +# +# 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 json + +import httplib2 + +FAKE_AUTH_URL = 'http://fake_uri.com/auth' + +TOKEN = "fake_token" +ALT_TOKEN = "alt_fake_token" + +# Fake Identity v2 constants +COMPUTE_ENDPOINTS_V2 = { + "endpoints": [ + { + "adminURL": "http://fake_url/v2/first_endpoint/admin", + "region": "NoMatchRegion", + "internalURL": "http://fake_url/v2/first_endpoint/internal", + "publicURL": "http://fake_url/v2/first_endpoint/public" + }, + { + "adminURL": "http://fake_url/v2/second_endpoint/admin", + "region": "FakeRegion", + "internalURL": "http://fake_url/v2/second_endpoint/internal", + "publicURL": "http://fake_url/v2/second_endpoint/public" + }, + ], + "type": "compute", + "name": "nova" +} + +CATALOG_V2 = [COMPUTE_ENDPOINTS_V2, ] + +ALT_IDENTITY_V2_RESPONSE = { + "access": { + "token": { + "expires": "2020-01-01T00:00:10Z", + "id": ALT_TOKEN, + "tenant": { + "id": "fake_tenant_id" + }, + }, + "user": { + "id": "fake_user_id", + }, + "serviceCatalog": CATALOG_V2, + }, +} + +IDENTITY_V2_RESPONSE = { + "access": { + "token": { + "expires": "2020-01-01T00:00:10Z", + "id": TOKEN, + "tenant": { + "id": "fake_tenant_id" + }, + }, + "user": { + "id": "fake_user_id", + }, + "serviceCatalog": CATALOG_V2, + }, +} + +# Fake Identity V3 constants +COMPUTE_ENDPOINTS_V3 = { + "endpoints": [ + { + "id": "first_compute_fake_service", + "interface": "public", + "region": "NoMatchRegion", + "url": "http://fake_url/v3/first_endpoint/api" + }, + { + "id": "second_fake_service", + "interface": "public", + "region": "FakeRegion", + "url": "http://fake_url/v3/second_endpoint/api" + }, + { + "id": "third_fake_service", + "interface": "admin", + "region": "MiddleEarthRegion", + "url": "http://fake_url/v3/third_endpoint/api" + } + + ], + "type": "compute", + "id": "fake_compute_endpoint" +} + +CATALOG_V3 = [COMPUTE_ENDPOINTS_V3, ] + +IDENTITY_V3_RESPONSE = { + "token": { + "methods": [ + "token", + "password" + ], + "expires_at": "2020-01-01T00:00:10.000123Z", + "project": { + "domain": { + "id": "fake_domain_id", + "name": "fake" + }, + "id": "project_id", + "name": "project_name" + }, + "user": { + "domain": { + "id": "fake_domain_id", + "name": "domain_name" + }, + "id": "fake_user_id", + "name": "username" + }, + "issued_at": "2013-05-29T16:55:21.468960Z", + "catalog": CATALOG_V3 + } +} + +ALT_IDENTITY_V3 = IDENTITY_V3_RESPONSE + + +def _fake_v3_response(self, uri, method="GET", body=None, headers=None, + redirections=5, connection_type=None): + fake_headers = { + "status": "201", + "x-subject-token": TOKEN + } + return (httplib2.Response(fake_headers), + json.dumps(IDENTITY_V3_RESPONSE)) + + +def _fake_v2_response(self, uri, method="GET", body=None, headers=None, + redirections=5, connection_type=None): + return (httplib2.Response({"status": "200"}), + json.dumps(IDENTITY_V2_RESPONSE)) + + +def _fake_auth_failure_response(): + # the response body isn't really used in this case, but lets send it anyway + # to have a safe check in some future change on the rest client. + body = { + "unauthorized": { + "message": "Unauthorized", + "code": "401" + } + } + return httplib2.Response({"status": "401"}), json.dumps(body) diff --git a/tempest_lib/tests/test_auth.py b/tempest_lib/tests/test_auth.py new file mode 100644 index 0000000..6249291 --- /dev/null +++ b/tempest_lib/tests/test_auth.py @@ -0,0 +1,407 @@ +# Copyright 2014 IBM Corp. +# All Rights Reserved. +# +# 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 copy +import datetime + +from oslotest import mockpatch + +from tempest_lib import auth +from tempest_lib import exceptions +from tempest_lib.services.identity.v2 import token_client as v2_client +from tempest_lib.services.identity.v3 import token_client as v3_client +from tempest_lib.tests import base +from tempest_lib.tests import fake_credentials +from tempest_lib.tests import fake_http +from tempest_lib.tests import fake_identity + + +def fake_get_credentials(fill_in=True, identity_version='v2', **kwargs): + return fake_credentials.FakeCredentials() + + +class BaseAuthTestsSetUp(base.TestCase): + _auth_provider_class = None + credentials = fake_credentials.FakeCredentials() + + def _auth(self, credentials, auth_url, **params): + """returns auth method according to keystone""" + return self._auth_provider_class(credentials, auth_url, **params) + + def setUp(self): + super(BaseAuthTestsSetUp, self).setUp() + self.fake_http = fake_http.fake_httplib2(return_type=200) + self.stubs.Set(auth, 'get_credentials', fake_get_credentials) + self.auth_provider = self._auth(self.credentials, + fake_identity.FAKE_AUTH_URL) + + +class TestBaseAuthProvider(BaseAuthTestsSetUp): + """Tests for base AuthProvider + + This tests auth.AuthProvider class which is base for the other so we + obviously don't test not implemented method or the ones which strongly + depends on them. + """ + + class FakeAuthProviderImpl(auth.AuthProvider): + def _decorate_request(self): + pass + + def _fill_credentials(self): + pass + + def _get_auth(self): + pass + + def base_url(self): + pass + + def is_expired(self): + pass + + _auth_provider_class = FakeAuthProviderImpl + + def _auth(self, credentials, auth_url, **params): + """returns auth method according to keystone""" + return self._auth_provider_class(credentials, **params) + + def test_check_credentials_bad_type(self): + self.assertFalse(self.auth_provider.check_credentials([])) + + def test_auth_data_property_when_cache_exists(self): + self.auth_provider.cache = 'foo' + self.useFixture(mockpatch.PatchObject(self.auth_provider, + 'is_expired', + return_value=False)) + self.assertEqual('foo', getattr(self.auth_provider, 'auth_data')) + + def test_delete_auth_data_property_through_deleter(self): + self.auth_provider.cache = 'foo' + del self.auth_provider.auth_data + self.assertIsNone(self.auth_provider.cache) + + def test_delete_auth_data_property_through_clear_auth(self): + self.auth_provider.cache = 'foo' + self.auth_provider.clear_auth() + self.assertIsNone(self.auth_provider.cache) + + def test_set_and_reset_alt_auth_data(self): + self.auth_provider.set_alt_auth_data('foo', 'bar') + self.assertEqual(self.auth_provider.alt_part, 'foo') + self.assertEqual(self.auth_provider.alt_auth_data, 'bar') + + self.auth_provider.reset_alt_auth_data() + self.assertIsNone(self.auth_provider.alt_part) + self.assertIsNone(self.auth_provider.alt_auth_data) + + def test_auth_class(self): + self.assertRaises(TypeError, + auth.AuthProvider, + fake_credentials.FakeCredentials) + + +class TestKeystoneV2AuthProvider(BaseAuthTestsSetUp): + _endpoints = fake_identity.IDENTITY_V2_RESPONSE['access']['serviceCatalog'] + _auth_provider_class = auth.KeystoneV2AuthProvider + credentials = fake_credentials.FakeKeystoneV2Credentials() + + def setUp(self): + super(TestKeystoneV2AuthProvider, self).setUp() + self.stubs.Set(v2_client.TokenClientJSON, 'raw_request', + fake_identity._fake_v2_response) + self.target_url = 'test_api' + + def _get_fake_alt_identity(self): + return fake_identity.ALT_IDENTITY_V2_RESPONSE['access'] + + def _get_result_url_from_endpoint(self, ep, endpoint_type='publicURL', + replacement=None): + if replacement: + return ep[endpoint_type].replace('v2', replacement) + return ep[endpoint_type] + + def _get_token_from_fake_identity(self): + return fake_identity.TOKEN + + def _get_from_fake_identity(self, attr): + access = fake_identity.IDENTITY_V2_RESPONSE['access'] + if attr == 'user_id': + return access['user']['id'] + elif attr == 'tenant_id': + return access['token']['tenant']['id'] + + def _test_request_helper(self, filters, expected): + url, headers, body = self.auth_provider.auth_request('GET', + self.target_url, + filters=filters) + + self.assertEqual(expected['url'], url) + self.assertEqual(expected['token'], headers['X-Auth-Token']) + self.assertEqual(expected['body'], body) + + def _auth_data_with_expiry(self, date_as_string): + token, access = self.auth_provider.auth_data + access['token']['expires'] = date_as_string + return token, access + + def test_request(self): + filters = { + 'service': 'compute', + 'endpoint_type': 'publicURL', + 'region': 'FakeRegion' + } + + url = self._get_result_url_from_endpoint( + self._endpoints[0]['endpoints'][1]) + '/' + self.target_url + + expected = { + 'body': None, + 'url': url, + 'token': self._get_token_from_fake_identity(), + } + self._test_request_helper(filters, expected) + + def test_request_with_alt_auth_cleans_alt(self): + self.auth_provider.set_alt_auth_data( + 'body', + (fake_identity.ALT_TOKEN, self._get_fake_alt_identity())) + self.test_request() + # Assert alt auth data is clear after it + self.assertIsNone(self.auth_provider.alt_part) + self.assertIsNone(self.auth_provider.alt_auth_data) + + def test_request_with_alt_part_without_alt_data(self): + """Test empty alternate auth data + + Assert that when alt_part is defined, the corresponding original + request element is kept the same. + """ + filters = { + 'service': 'compute', + 'endpoint_type': 'publicURL', + 'region': 'fakeRegion' + } + self.auth_provider.set_alt_auth_data('url', None) + + url, headers, body = self.auth_provider.auth_request('GET', + self.target_url, + filters=filters) + + self.assertEqual(url, self.target_url) + self.assertEqual(self._get_token_from_fake_identity(), + headers['X-Auth-Token']) + self.assertEqual(body, None) + + def test_request_with_bad_service(self): + filters = { + 'service': 'BAD_SERVICE', + 'endpoint_type': 'publicURL', + 'region': 'fakeRegion' + } + self.assertRaises(exceptions.EndpointNotFound, + self.auth_provider.auth_request, 'GET', + self.target_url, filters=filters) + + def test_request_without_service(self): + filters = { + 'service': None, + 'endpoint_type': 'publicURL', + 'region': 'fakeRegion' + } + self.assertRaises(exceptions.EndpointNotFound, + self.auth_provider.auth_request, 'GET', + self.target_url, filters=filters) + + def test_check_credentials_missing_attribute(self): + for attr in ['username', 'password']: + cred = copy.copy(self.credentials) + del cred[attr] + self.assertFalse(self.auth_provider.check_credentials(cred)) + + def test_fill_credentials(self): + self.auth_provider.fill_credentials() + creds = self.auth_provider.credentials + for attr in ['user_id', 'tenant_id']: + self.assertEqual(self._get_from_fake_identity(attr), + getattr(creds, attr)) + + def _test_base_url_helper(self, expected_url, filters, + auth_data=None): + + url = self.auth_provider.base_url(filters, auth_data) + self.assertEqual(url, expected_url) + + def test_base_url(self): + self.filters = { + 'service': 'compute', + 'endpoint_type': 'publicURL', + 'region': 'FakeRegion' + } + expected = self._get_result_url_from_endpoint( + self._endpoints[0]['endpoints'][1]) + self._test_base_url_helper(expected, self.filters) + + def test_base_url_to_get_admin_endpoint(self): + self.filters = { + 'service': 'compute', + 'endpoint_type': 'adminURL', + 'region': 'FakeRegion' + } + expected = self._get_result_url_from_endpoint( + self._endpoints[0]['endpoints'][1], endpoint_type='adminURL') + self._test_base_url_helper(expected, self.filters) + + def test_base_url_unknown_region(self): + """If the region is unknown, the first endpoint is returned.""" + self.filters = { + 'service': 'compute', + 'endpoint_type': 'publicURL', + 'region': 'AintNoBodyKnowThisRegion' + } + expected = self._get_result_url_from_endpoint( + self._endpoints[0]['endpoints'][0]) + self._test_base_url_helper(expected, self.filters) + + def test_base_url_with_non_existent_service(self): + self.filters = { + 'service': 'BAD_SERVICE', + 'endpoint_type': 'publicURL', + 'region': 'FakeRegion' + } + self.assertRaises(exceptions.EndpointNotFound, + self._test_base_url_helper, None, self.filters) + + def test_base_url_without_service(self): + self.filters = { + 'endpoint_type': 'publicURL', + 'region': 'FakeRegion' + } + self.assertRaises(exceptions.EndpointNotFound, + self._test_base_url_helper, None, self.filters) + + def test_base_url_with_api_version_filter(self): + self.filters = { + 'service': 'compute', + 'endpoint_type': 'publicURL', + 'region': 'FakeRegion', + 'api_version': 'v12' + } + expected = self._get_result_url_from_endpoint( + self._endpoints[0]['endpoints'][1], replacement='v12') + self._test_base_url_helper(expected, self.filters) + + def test_base_url_with_skip_path_filter(self): + self.filters = { + 'service': 'compute', + 'endpoint_type': 'publicURL', + 'region': 'FakeRegion', + 'skip_path': True + } + expected = 'http://fake_url/' + self._test_base_url_helper(expected, self.filters) + + def test_token_not_expired(self): + expiry_data = datetime.datetime.utcnow() + datetime.timedelta(days=1) + auth_data = self._auth_data_with_expiry( + expiry_data.strftime(self.auth_provider.EXPIRY_DATE_FORMAT)) + self.assertFalse(self.auth_provider.is_expired(auth_data)) + + def test_token_expired(self): + expiry_data = datetime.datetime.utcnow() - datetime.timedelta(hours=1) + auth_data = self._auth_data_with_expiry( + expiry_data.strftime(self.auth_provider.EXPIRY_DATE_FORMAT)) + self.assertTrue(self.auth_provider.is_expired(auth_data)) + + def test_token_not_expired_to_be_renewed(self): + expiry_data = (datetime.datetime.utcnow() + + self.auth_provider.token_expiry_threshold / 2) + auth_data = self._auth_data_with_expiry( + expiry_data.strftime(self.auth_provider.EXPIRY_DATE_FORMAT)) + self.assertTrue(self.auth_provider.is_expired(auth_data)) + + +class TestKeystoneV3AuthProvider(TestKeystoneV2AuthProvider): + _endpoints = fake_identity.IDENTITY_V3_RESPONSE['token']['catalog'] + _auth_provider_class = auth.KeystoneV3AuthProvider + credentials = fake_credentials.FakeKeystoneV3Credentials() + + def setUp(self): + super(TestKeystoneV3AuthProvider, self).setUp() + self.stubs.Set(v3_client.V3TokenClientJSON, 'raw_request', + fake_identity._fake_v3_response) + + def _get_fake_alt_identity(self): + return fake_identity.ALT_IDENTITY_V3['token'] + + def _get_result_url_from_endpoint(self, ep, replacement=None): + if replacement: + return ep['url'].replace('v3', replacement) + return ep['url'] + + def _auth_data_with_expiry(self, date_as_string): + token, access = self.auth_provider.auth_data + access['expires_at'] = date_as_string + return token, access + + def _get_from_fake_identity(self, attr): + token = fake_identity.IDENTITY_V3_RESPONSE['token'] + if attr == 'user_id': + return token['user']['id'] + elif attr == 'project_id': + return token['project']['id'] + elif attr == 'user_domain_id': + return token['user']['domain']['id'] + elif attr == 'project_domain_id': + return token['project']['domain']['id'] + + def test_check_credentials_missing_attribute(self): + # reset credentials to fresh ones + self.credentials.reset() + for attr in ['username', 'password', 'user_domain_name', + 'project_domain_name']: + cred = copy.copy(self.credentials) + del cred[attr] + self.assertFalse(self.auth_provider.check_credentials(cred), + "Credentials should be invalid without %s" % attr) + + def test_check_domain_credentials_missing_attribute(self): + # reset credentials to fresh ones + self.credentials.reset() + domain_creds = fake_credentials.FakeKeystoneV3DomainCredentials() + for attr in ['username', 'password', 'user_domain_name']: + cred = copy.copy(domain_creds) + del cred[attr] + self.assertFalse(self.auth_provider.check_credentials(cred), + "Credentials should be invalid without %s" % attr) + + def test_fill_credentials(self): + self.auth_provider.fill_credentials() + creds = self.auth_provider.credentials + for attr in ['user_id', 'project_id', 'user_domain_id', + 'project_domain_id']: + self.assertEqual(self._get_from_fake_identity(attr), + getattr(creds, attr)) + + # Overwrites v2 test + def test_base_url_to_get_admin_endpoint(self): + self.filters = { + 'service': 'compute', + 'endpoint_type': 'admin', + 'region': 'MiddleEarthRegion' + } + expected = self._get_result_url_from_endpoint( + self._endpoints[0]['endpoints'][2]) + self._test_base_url_helper(expected, self.filters)