openstack/python-keystoneclient
Code Issues Proposed changes

Create V2 Auth Plugins

Browse Source 
Extract the authentication code from a v2 client and move it to a series
of auth plugins. Auth plugins each represent one method of
authenticating with a server and there is a factory method on the base
class to select the appropriate plugin from a group of arguments.

When a v2 client wants to do authentication it will create
a new v2 auth plugin, do the authentication and then take that result
for the client to use.

Change-Id: I4dd7474643ed5c2a3204ea2ec56029f926010c2c
blueprint: auth-plugins
This commit is contained in:
Jamie Lennox
2014-01-21 11:40:16 +10:00
parent bbd30eee3f
commit 96b8e81b1d
6 changed files with 314 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
keystoneclient/auth/identity/base.py
View File

@@ -22,6 +22,9 @@ LOG = logging.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class BaseIdentityPlugin(base.BaseAuthPlugin):
# we count a token as valid if it is valid for at least this many seconds
MIN_TOKEN_LIFE_SECONDS = 1
def __init__(self,
auth_url=None,
username=None,
@@ -32,13 +35,15 @@ class BaseIdentityPlugin(base.BaseAuthPlugin):
super(BaseIdentityPlugin, self).__init__()
self.auth_url = auth_url
self.auth_ref = None
# NOTE(jamielennox): DEPRECATED. The following should not really be set
# here but handled by the individual auth plugin.
self.username = username
self.password = password
self.token = token
self.trust_id = trust_id
self.auth_ref = None
@abc.abstractmethod
def get_auth_ref(self, session, **kwargs):
"""Obtain a token from an OpenStack Identity Service.
@@ -48,11 +53,40 @@ class BaseIdentityPlugin(base.BaseAuthPlugin):
This function should not be called independently and is expected to be
invoked via the do_authenticate function.
This function will be invoked if the AcessInfo object cached by the
plugin is not valid. Thus plugins should always fetch a new AccessInfo
when invoked. If you are looking to just retrieve the current auth
data then you should use get_access.
:raises HTTPError: An error from an invalid HTTP response.
:returns AccessInfo: Token access information.
"""
def get_token(self, session, **kwargs):
if not self.auth_ref or self.auth_ref.will_expire_soon(1):
"""Return a valid auth token.
If a valid token is not present then a new one will be fetched using
the session and kwargs.
:raises HTTPError: An error from an invalid HTTP response.
:return string: A valid token.
"""
return self.get_access(session, **kwargs).auth_token
def get_access(self, session, **kwargs):
"""Fetch or return a current AccessInfo object.
If a valid AccessInfo is present then it is returned otherwise kwargs
and session are used to fetch a new one.
:raises HTTPError: An error from an invalid HTTP response.
:returns AccessInfo: Valid AccessInfo
"""
if (not self.auth_ref or
self.auth_ref.will_expire_soon(self.MIN_TOKEN_LIFE_SECONDS)):
self.auth_ref = self.get_auth_ref(session, **kwargs)
return self.auth_ref.auth_token
return self.auth_ref

129
keystoneclient/auth/identity/v2.py Normal file
View File

@@ -0,0 +1,129 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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 six
from keystoneclient import access
from keystoneclient.auth.identity import base
from keystoneclient import exceptions
@six.add_metaclass(abc.ABCMeta)
class Auth(base.BaseIdentityPlugin):
@staticmethod
def factory(auth_url, **kwargs):
"""Construct a plugin appropriate to your available arguments.
This function should only be used for loading authentication from a
config file or other source where you do not know the type of plugin
that is required.
If you know the style of authorization you require then you should
construct that plugin directly.
:raises NoMatchingPlugin: if a plugin cannot be constructed.
return Auth: a plugin that can be passed to a session.
"""
username = kwargs.pop('username', None)
password = kwargs.pop('password', None)
token = kwargs.pop('token', None)
if token:
return Token(auth_url, token, **kwargs)
elif username and password:
return Password(auth_url, username, password, **kwargs)
msg = 'A username and password or token is required.'
raise exceptions.NoMatchingPlugin(msg)
def __init__(self, auth_url,
trust_id=None,
tenant_id=None,
tenant_name=None):
"""Construct an Identity V2 Authentication Plugin.
:param string auth_url: Identity service endpoint for authorization.
:param string trust_id: Trust ID for trust scoping.
:param string tenant_id: Tenant ID for project scoping.
:param string tenant_name: Tenant name for project scoping.
"""
super(Auth, self).__init__(auth_url=auth_url)
self.trust_id = trust_id
self.tenant_id = tenant_id
self.tenant_name = tenant_name
def get_auth_ref(self, session, **kwargs):
headers = {}
url = self.auth_url + '/tokens'
params = {'auth': self.get_auth_data(headers)}
if self.tenant_id:
params['auth']['tenantId'] = self.tenant_id
elif self.tenant_name:
params['auth']['tenantName'] = self.tenant_name
if self.trust_id:
params['auth']['trust_id'] = self.trust_id
resp = session.post(url, json=params, headers=headers,
authenticated=False)
return access.AccessInfoV2(**resp.json()['access'])
@abc.abstractmethod
def get_auth_data(self, headers=None):
"""Return the authentication section of an auth plugin.
:param dict headers: The headers that will be sent with the auth
request if a plugin needs to add to them.
:return dict: A dict of authentication data for the auth type.
"""
class Password(Auth):
def __init__(self, auth_url, username, password, **kwargs):
"""A plugin for authenticating with a username and password.
:param string auth_url: Identity service endpoint for authorization.
:param string username: Username for authentication.
:param string password: Password for authentication.
"""
super(Password, self).__init__(auth_url, **kwargs)
self.username = username
self.password = password
def get_auth_data(self, headers=None):
return {'passwordCredentials': {'username': self.username,
'password': self.password}}
class Token(Auth):
def __init__(self, auth_url, token, **kwargs):
"""A plugin for authenticating with an existing token.
:param string auth_url: Identity service endpoint for authorization.
:param string token: Existing token for authentication.
"""
super(Token, self).__init__(auth_url, **kwargs)
self.token = token
def get_auth_data(self, headers=None):
if headers is not None:
headers['X-Auth-Token'] = self.token
return {'token': {'id': self.token}}

5
keystoneclient/exceptions.py
View File

@@ -51,3 +51,8 @@ class VersionNotAvailable(DiscoveryFailure):
class MissingAuthPlugin(ClientException):
"""An authenticated request is required but no plugin available."""
class NoMatchingPlugin(ClientException):
"""There were no auth plugins that could be created from the parameters
provided."""

13
keystoneclient/session.py
View File

@@ -314,8 +314,17 @@ class Session(object):
user_agent=kwargs.pop('user_agent', None))
def get_token(self):
"""Return a token as provided by the auth plugin."""
"""Return a token as provided by the auth plugin.
:raises AuthorizationFailure: if a new token fetch fails.
:returns string: A valid token.
"""
if not self.auth:
raise exceptions.MissingAuthPlugin("Token Required")
return self.auth.get_token(self)
try:
return self.auth.get_token(self)
except exceptions.HTTPError as exc:
raise exceptions.AuthorizationFailure("Authentication failure: "
"%s" % exc)

114
keystoneclient/tests/auth/test_identity_v2.py Normal file
View File

@@ -0,0 +1,114 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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 httpretty
from keystoneclient.auth.identity import v2
from keystoneclient import exceptions
from keystoneclient import session
from keystoneclient.tests import utils
class V2IdentityPlugin(utils.TestCase):
TEST_ROOT_URL = 'http://127.0.0.1:5000/'
TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v2.0')
TEST_ROOT_ADMIN_URL = 'http://127.0.0.1:35357/'
TEST_ADMIN_URL = '%s%s' % (TEST_ROOT_ADMIN_URL, 'v2.0')
TEST_PASS = 'password'
TEST_SERVICE_CATALOG = []
def setUp(self):
super(V2IdentityPlugin, self).setUp()
self.TEST_RESPONSE_DICT = {
"access": {
"token": {
"expires": "2020-01-01T00:00:10.000123Z",
"id": self.TEST_TOKEN,
"tenant": {
"id": self.TEST_TENANT_ID
},
},
"user": {
"id": self.TEST_USER
},
"serviceCatalog": self.TEST_SERVICE_CATALOG,
},
}
def _plugin(self, auth_url=TEST_URL, **kwargs):
return v2.Auth.factory(auth_url, **kwargs)
def _session(self, **kwargs):
return session.Session(auth=self._plugin(**kwargs))
def stub_auth(self, **kwargs):
self.stub_url(httpretty.POST, ['tokens'], **kwargs)
@httpretty.activate
def test_authenticate_with_username_password(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
s = self._session(username=self.TEST_USER, password=self.TEST_PASS)
self.assertIsInstance(s.auth, v2.Password)
s.get_token()
req = {'auth': {'passwordCredentials': {'username': self.TEST_USER,
'password': self.TEST_PASS}}}
self.assertRequestBodyIs(json=req)
self.assertEqual(s.auth.auth_ref.auth_token, self.TEST_TOKEN)
@httpretty.activate
def test_authenticate_with_username_password_scoped(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
s = self._session(username=self.TEST_USER, password=self.TEST_PASS,
tenant_id=self.TEST_TENANT_ID)
self.assertIsInstance(s.auth, v2.Password)
s.get_token()
req = {'auth': {'passwordCredentials': {'username': self.TEST_USER,
'password': self.TEST_PASS},
'tenantId': self.TEST_TENANT_ID}}
self.assertRequestBodyIs(json=req)
self.assertEqual(s.auth.auth_ref.auth_token, self.TEST_TOKEN)
@httpretty.activate
def test_authenticate_with_token(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
s = self._session(token='foo')
self.assertIsInstance(s.auth, v2.Token)
s.get_token()
req = {'auth': {'token': {'id': 'foo'}}}
self.assertRequestBodyIs(json=req)
self.assertRequestHeaderEqual('x-Auth-Token', 'foo')
self.assertEqual(s.auth.auth_ref.auth_token, self.TEST_TOKEN)
def test_missing_auth_params(self):
self.assertRaises(exceptions.NoMatchingPlugin, self._plugin)
@httpretty.activate
def test_with_trust_id(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
s = self._session(username=self.TEST_USER, password=self.TEST_PASS,
trust_id='trust')
s.get_token()
req = {'auth': {'passwordCredentials': {'username': self.TEST_USER,
'password': self.TEST_PASS},
'trust_id': 'trust'}}
self.assertRequestBodyIs(json=req)
self.assertEqual(s.auth.auth_ref.auth_token, self.TEST_TOKEN)

58
keystoneclient/v2_0/client.py
View File

@@ -15,6 +15,7 @@
import logging
from keystoneclient.auth.identity import v2 as v2_auth
from keystoneclient import exceptions
from keystoneclient import httpclient
from keystoneclient.v2_0 import ec2
@@ -137,7 +138,9 @@ class Client(httpclient.HTTPClient):
# extensions
self.ec2 = ec2.CredentialsManager(self)
if self.management_url is None:
# DEPRECATED: if session is passed then we go to the new behaviour of
# authenticating on the first required call.
if not kwargs.get('session') and self.management_url is None:
self.authenticate()
def get_raw_token_from_identity_service(self, auth_url, username=None,
@@ -148,53 +151,26 @@ class Client(httpclient.HTTPClient):
**kwargs):
"""Authenticate against the v2 Identity API.
:returns: (``resp``, ``body``) if authentication was successful.
:returns: access.AccessInfo if authentication was successful.
:raises: AuthorizationFailure if unable to authenticate or validate
the existing authorization token
:raises: ValueError if insufficient parameters are used.
"""
try:
return self._base_authN(auth_url,
username=username,
tenant_id=project_id or tenant_id,
tenant_name=project_name or tenant_name,
password=password,
trust_id=trust_id,
token=token)
if auth_url is None:
raise ValueError("Cannot authenticate without an auth_url")
a = v2_auth.Auth.factory(auth_url,
username=username,
password=password,
token=token,
trust_id=trust_id,
tenant_id=project_id or tenant_id,
tenant_name=project_name or tenant_name)
return a.get_auth_ref(self.session)
except (exceptions.AuthorizationFailure, exceptions.Unauthorized):
_logger.debug("Authorization Failed.")
raise
except Exception as e:
raise exceptions.AuthorizationFailure("Authorization Failed: "
"%s" % e)
def _base_authN(self, auth_url, username=None, password=None,
tenant_name=None, tenant_id=None, trust_id=None,
token=None):
"""Takes a username, password, and optionally a tenant_id or
tenant_name to get an authentication token from keystone.
May also take a token and a tenant_id to re-scope a token
to a tenant, or a token, tenant_id and trust_id and re-scope
the token to the trust
"""
headers = {}
if auth_url is None:
raise ValueError("Cannot authenticate without a valid auth_url")
url = auth_url + "/tokens"
if token:
headers['X-Auth-Token'] = token
params = {"auth": {"token": {"id": token}}}
elif username and password:
params = {"auth": {"passwordCredentials": {"username": username,
"password": password}}}
else:
raise ValueError('A username and password or token is required.')
if tenant_id:
params['auth']['tenantId'] = tenant_id
elif tenant_name:
params['auth']['tenantName'] = tenant_name
if trust_id:
params['auth']['trust_id'] = trust_id
resp, body = self.request(url, 'POST', body=params, headers=headers)
return resp, body