openstack/python-keystoneclient
Code Issues Proposed changes

Merge "Add get_headers interface to authentication plugins"

Browse Source
This commit is contained in:
Jenkins 2015-02-09 05:54:33 +00:00 committed by Gerrit Code Review
commit 4ee6e3302a
6 changed files with 180 additions and 32 deletions

1
keystoneclient/auth/__init__.py

@ -21,6 +21,7 @@ __all__ = [
'AUTH_INTERFACE',
'BaseAuthPlugin',
'get_plugin_class',
'IDENTITY_AUTH_HEADER_NAME',
'PLUGIN_NAMESPACE',
# auth.cli

51
keystoneclient/auth/base.py

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import abc
import os
import six
@ -25,6 +24,7 @@ from keystoneclient import exceptions
AUTH_INTERFACE = object()
PLUGIN_NAMESPACE = 'keystoneclient.auth.plugin'
IDENTITY_AUTH_HEADER_NAME = 'X-Auth-Token'
def get_plugin_class(name):
@ -48,11 +48,9 @@ def get_plugin_class(name):
return mgr.driver
@six.add_metaclass(abc.ABCMeta)
class BaseAuthPlugin(object):
"""The basic structure of an authentication plugin."""
@abc.abstractmethod
def get_token(self, session, **kwargs):
"""Obtain a token.
@ -65,6 +63,15 @@ class BaseAuthPlugin(object):
Returning None will indicate that no token was able to be retrieved.
This function is misplaced as it should only be required for auth
plugins that use the 'X-Auth-Token' header. However due to the way
plugins evolved this method is required and often called to trigger an
authentication request on a new plugin.
When implementing a new plugin it is advised that you implement this
method, however if you don't require the 'X-Auth-Token' header override
the `get_headers` method instead.
:param session: A session object so the plugin can make HTTP calls.
:type session: keystoneclient.session.Session
@ -72,6 +79,44 @@ class BaseAuthPlugin(object):
:rtype: string
"""
def get_headers(self, session, **kwargs):
"""Fetch authentication headers for message.
This is a more generalized replacement of the older get_token to allow
plugins to specify different or additional authentication headers to
the OpenStack standard 'X-Auth-Token' header.
How the authentication headers are obtained is up to the plugin. If the
headers are still valid they may be re-used, retrieved from cache or
the plugin may invoke an authentication request against a server.
The default implementation of get_headers calls the `get_token` method
to enable older style plugins to continue functioning unchanged.
Subclasses should feel free to completely override this function to
provide the headers that they want.
There are no required kwargs. They are passed directly to the auth
plugin and they are implementation specific.
Returning None will indicate that no token was able to be retrieved and
that authorization was a failure. Adding no authentication data can be
achieved by returning an empty dictionary.
:param session: The session object that the auth_plugin belongs to.
:type session: keystoneclient.session.Session
:returns: Headers that are set to authenticate a message or None for
failure. Note that when checking this value that the empty
dict is a valid, non-failure response.
:rtype: dict
"""
token = self.get_token(session)
if not token:
return None
return {IDENTITY_AUTH_HEADER_NAME: token}
def get_endpoint(self, session, **kwargs):
"""Return an endpoint for the client.

46
keystoneclient/session.py

@ -299,12 +299,13 @@ class Session(object):
authenticated = bool(auth or self.auth)
if authenticated:
token = self.get_token(auth)
auth_headers = self.get_auth_headers(auth)
if not token:
raise exceptions.AuthorizationFailure(_("No token Available"))
if auth_headers is None:
msg = _('No valid authentication is available')
raise exceptions.AuthorizationFailure(msg)
headers['X-Auth-Token'] = token
headers.update(auth_headers)
if osprofiler_web:
headers.update(osprofiler_web.get_trace_id_headers())
@ -371,9 +372,10 @@ class Session(object):
# and then retrying the request. This is only tried once.
if resp.status_code == 401 and authenticated and allow_reauth:
if self.invalidate(auth):
token = self.get_token(auth)
if token:
headers['X-Auth-Token'] = token
auth_headers = self.get_auth_headers(auth)
if auth_headers is not None:
headers.update(auth_headers)
resp = send(**kwargs)
if raise_exc and resp.status_code >= 400:
@ -563,6 +565,24 @@ class Session(object):
return auth
def get_auth_headers(self, auth=None, **kwargs):
"""Return auth headers as provided by the auth plugin.
:param auth: The auth plugin to use for token. Overrides the plugin
on the session. (optional)
:type auth: :py:class:`keystoneclient.auth.base.BaseAuthPlugin`
:raises keystoneclient.exceptions.AuthorizationFailure: if a new token
fetch fails.
:raises keystoneclient.exceptions.MissingAuthPlugin: if a plugin is not
available.
:returns: Authentication headers or None for failure.
:rtype: dict
"""
auth = self._auth_required(auth, 'fetch a token')
return auth.get_headers(self, **kwargs)
def get_token(self, auth=None):
"""Return a token as provided by the auth plugin.
@ -575,16 +595,14 @@ class Session(object):
:raises keystoneclient.exceptions.MissingAuthPlugin: if a plugin is not
available.
*DEPRECATED*: This assumes that the only header that is used to
authenticate a message is 'X-Auth-Token'. This may not be
correct. Use get_auth_headers instead.
:returns: A valid token.
:rtype: string
"""
auth = self._auth_required(auth, 'fetch a token')
try:
return auth.get_token(self)
except exceptions.HttpError as exc:
raise exceptions.AuthorizationFailure(
_("Authentication failure: %s") % exc)
return (self.get_auth_headers(auth) or {}).get('X-Auth-Token')
def get_endpoint(self, auth=None, **kwargs):
"""Get an endpoint as provided by the auth plugin.

55
keystoneclient/tests/auth/test_identity_common.py

@ -221,7 +221,7 @@ class CommonIdentityTests(object):
s = session.Session(auth=a)
# trigger token fetching
s.get_token()
s.get_auth_headers()
self.assertTrue(a.auth_ref)
self.assertTrue(a.invalidate())
@ -368,3 +368,56 @@ class CatalogHackTests(utils.TestCase):
version=(3, 0))
self.assertEqual(self.V2_URL, endpoint)
class GenericPlugin(base.BaseAuthPlugin):
BAD_TOKEN = uuid.uuid4().hex
def __init__(self):
super(GenericPlugin, self).__init__()
self.endpoint = 'http://keystone.host:5000'
self.headers = {'headerA': 'valueA',
'headerB': 'valueB'}
def url(self, prefix):
return '%s/%s' % (self.endpoint, prefix)
def get_token(self, session, **kwargs):
# NOTE(jamielennox): by specifying get_headers this should not be used
return self.BAD_TOKEN
def get_headers(self, session, **kwargs):
return self.headers
def get_endpoint(self, session, **kwargs):
return self.endpoint
class GenericAuthPluginTests(utils.TestCase):
# filter doesn't matter to GenericPlugin, but we have to specify one
ENDPOINT_FILTER = {uuid.uuid4().hex: uuid.uuid4().hex}
def setUp(self):
super(GenericAuthPluginTests, self).setUp()
self.auth = GenericPlugin()
self.session = session.Session(auth=self.auth)
def test_setting_headers(self):
text = uuid.uuid4().hex
self.stub_url('GET', base_url=self.auth.url('prefix'), text=text)
resp = self.session.get('prefix', endpoint_filter=self.ENDPOINT_FILTER)
self.assertEqual(text, resp.text)
for k, v in six.iteritems(self.auth.headers):
self.assertRequestHeaderEqual(k, v)
self.assertIsNone(self.session.get_token())
self.assertEqual(self.auth.headers,
self.session.get_auth_headers())
self.assertNotIn('X-Auth-Token', self.requests.last_request.headers)

23
keystoneclient/tests/auth/test_identity_v2.py

@ -102,7 +102,8 @@ class V2IdentityPlugin(utils.TestCase):
password=self.TEST_PASS)
self.assertIsNone(a.user_id)
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'passwordCredentials': {'username': self.TEST_USER,
'password': self.TEST_PASS}}}
@ -117,7 +118,8 @@ class V2IdentityPlugin(utils.TestCase):
password=self.TEST_PASS)
self.assertIsNone(a.username)
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'passwordCredentials': {'userId': self.TEST_USER,
'password': self.TEST_PASS}}}
@ -132,7 +134,8 @@ class V2IdentityPlugin(utils.TestCase):
password=self.TEST_PASS, tenant_id=self.TEST_TENANT_ID)
self.assertIsNone(a.user_id)
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'passwordCredentials': {'username': self.TEST_USER,
'password': self.TEST_PASS},
@ -146,7 +149,8 @@ class V2IdentityPlugin(utils.TestCase):
password=self.TEST_PASS, tenant_id=self.TEST_TENANT_ID)
self.assertIsNone(a.username)
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'passwordCredentials': {'userId': self.TEST_USER,
'password': self.TEST_PASS},
@ -158,7 +162,8 @@ class V2IdentityPlugin(utils.TestCase):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
a = v2.Token(self.TEST_URL, 'foo')
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'token': {'id': 'foo'}}}
self.assertRequestBodyIs(json=req)
@ -172,7 +177,8 @@ class V2IdentityPlugin(utils.TestCase):
a = v2.Password(self.TEST_URL, username=self.TEST_USER,
password=self.TEST_PASS, trust_id='trust')
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'passwordCredentials': {'username': self.TEST_USER,
'password': self.TEST_PASS},
@ -266,8 +272,11 @@ class V2IdentityPlugin(utils.TestCase):
s = session.Session(auth=a)
self.assertEqual('token1', s.get_token())
self.assertEqual({'X-Auth-Token': 'token1'}, s.get_auth_headers())
a.invalidate()
self.assertEqual('token2', s.get_token())
self.assertEqual({'X-Auth-Token': 'token2'}, s.get_auth_headers())
def test_doesnt_log_password(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
@ -277,6 +286,8 @@ class V2IdentityPlugin(utils.TestCase):
password=password)
s = session.Session(auth=a)
self.assertEqual(self.TEST_TOKEN, s.get_token())
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
self.assertNotIn(password, self.logger.output)
def test_password_with_no_user_id_or_name(self):

36
keystoneclient/tests/auth/test_identity_v3.py

@ -185,7 +185,8 @@ class V3IdentityPlugin(utils.TestCase):
password=self.TEST_PASS)
s = session.Session(auth=a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['password'],
@ -225,7 +226,9 @@ class V3IdentityPlugin(utils.TestCase):
a = v3.Password(self.TEST_URL, username=self.TEST_USER,
password=self.TEST_PASS, domain_id=self.TEST_DOMAIN_ID)
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['password'],
@ -241,7 +244,9 @@ class V3IdentityPlugin(utils.TestCase):
password=self.TEST_PASS,
project_id=self.TEST_DOMAIN_ID)
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['password'],
@ -256,7 +261,9 @@ class V3IdentityPlugin(utils.TestCase):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
a = v3.Token(self.TEST_URL, self.TEST_TOKEN)
s = session.Session(auth=a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['token'],
@ -279,7 +286,8 @@ class V3IdentityPlugin(utils.TestCase):
a.auth_ref = access.AccessInfo.factory(body=d)
s = session.Session(auth=a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
self.assertEqual(a.auth_ref['expires_at'],
self.TEST_RESPONSE_DICT['token']['expires_at'])
@ -288,15 +296,20 @@ class V3IdentityPlugin(utils.TestCase):
a = v3.Password(self.TEST_URL, username='username',
password='password', project_id='project',
domain_id='domain')
self.assertRaises(exceptions.AuthorizationFailure,
a.get_token, None)
self.assertRaises(exceptions.AuthorizationFailure,
a.get_headers, None)
def test_with_trust_id(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
a = v3.Password(self.TEST_URL, username=self.TEST_USER,
password=self.TEST_PASS, trust_id='trust')
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['password'],
@ -312,7 +325,9 @@ class V3IdentityPlugin(utils.TestCase):
t = v3.TokenMethod(token='foo')
a = v3.Auth(self.TEST_URL, [p, t], trust_id='trust')
s = session.Session(a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['password', 'token'],
@ -331,7 +346,8 @@ class V3IdentityPlugin(utils.TestCase):
a = v3.Auth(self.TEST_URL, [p, t], trust_id='trust')
s = session.Session(auth=a)
s.get_token()
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
req = {'auth': {'identity':
{'methods': ['password', 'token'],
@ -438,8 +454,10 @@ class V3IdentityPlugin(utils.TestCase):
s = session.Session(auth=a)
self.assertEqual('token1', s.get_token())
self.assertEqual({'X-Auth-Token': 'token1'}, s.get_auth_headers())
a.invalidate()
self.assertEqual('token2', s.get_token())
self.assertEqual({'X-Auth-Token': 'token2'}, s.get_auth_headers())
def test_doesnt_log_password(self):
self.stub_auth(json=self.TEST_RESPONSE_DICT)
@ -449,6 +467,8 @@ class V3IdentityPlugin(utils.TestCase):
password=password)
s = session.Session(a)
self.assertEqual(self.TEST_TOKEN, s.get_token())
self.assertEqual({'X-Auth-Token': self.TEST_TOKEN},
s.get_auth_headers())
self.assertNotIn(password, self.logger.output)