Merge "Add a FederatedBase v3 plugin"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
b0dd0ce80d
|
|
@ -11,6 +11,7 @@
|
|||
# under the License.
|
||||
|
||||
from keystoneclient.auth.identity.v3.base import * # noqa
|
||||
from keystoneclient.auth.identity.v3.federated import * # noqa
|
||||
from keystoneclient.auth.identity.v3.password import * # noqa
|
||||
from keystoneclient.auth.identity.v3.token import * # noqa
|
||||
|
||||
|
|
@ -20,6 +21,8 @@ __all__ = ['Auth',
|
|||
'AuthMethod',
|
||||
'BaseAuth',
|
||||
|
||||
'FederatedBaseAuth',
|
||||
|
||||
'Password',
|
||||
'PasswordMethod',
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,111 @@
|
|||
# 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
|
||||
|
||||
from oslo_config import cfg
|
||||
import six
|
||||
|
||||
from keystoneclient.auth.identity.v3 import base
|
||||
from keystoneclient.auth.identity.v3 import token
|
||||
|
||||
__all__ = ['FederatedBaseAuth']
|
||||
|
||||
|
||||
@six.add_metaclass(abc.ABCMeta)
|
||||
class FederatedBaseAuth(base.BaseAuth):
|
||||
|
||||
rescoping_plugin = token.Token
|
||||
|
||||
def __init__(self, auth_url, identity_provider, protocol, **kwargs):
|
||||
"""Class constructor accepting following parameters:
|
||||
|
||||
:param auth_url: URL of the Identity Service
|
||||
:type auth_url: string
|
||||
:param identity_provider: name of the Identity Provider the client
|
||||
will authenticate against. This parameter
|
||||
will be used to build a dynamic URL used to
|
||||
obtain unscoped OpenStack token.
|
||||
:type identity_provider: string
|
||||
|
||||
"""
|
||||
super(FederatedBaseAuth, self).__init__(auth_url=auth_url, **kwargs)
|
||||
self.identity_provider = identity_provider
|
||||
self.protocol = protocol
|
||||
|
||||
@classmethod
|
||||
def get_options(cls):
|
||||
options = super(FederatedBaseAuth, cls).get_options()
|
||||
|
||||
options.extend([
|
||||
cfg.StrOpt('identity-provider',
|
||||
help="Identity Provider's name"),
|
||||
cfg.StrOpt('protocol',
|
||||
help='Protocol for federated plugin'),
|
||||
])
|
||||
|
||||
return options
|
||||
|
||||
@property
|
||||
def federated_token_url(self):
|
||||
"""Full URL where authorization data is sent."""
|
||||
values = {
|
||||
'host': self.auth_url.rstrip('/'),
|
||||
'identity_provider': self.identity_provider,
|
||||
'protocol': self.protocol
|
||||
}
|
||||
url = ("%(host)s/OS-FEDERATION/identity_providers/"
|
||||
"%(identity_provider)s/protocols/%(protocol)s/auth")
|
||||
url = url % values
|
||||
|
||||
return url
|
||||
|
||||
def _get_scoping_data(self):
|
||||
return {'trust_id': self.trust_id,
|
||||
'domain_id': self.domain_id,
|
||||
'domain_name': self.domain_name,
|
||||
'project_id': self.project_id,
|
||||
'project_name': self.project_name,
|
||||
'project_domain_id': self.project_domain_id,
|
||||
'project_domain_name': self.project_domain_name}
|
||||
|
||||
def get_auth_ref(self, session, **kwargs):
|
||||
"""Authenticate retrieve token information.
|
||||
|
||||
This is a multi-step process where a client does federated authn
|
||||
receives an unscoped token.
|
||||
|
||||
If an unscoped token is successfully received and scoping information
|
||||
is present then the token is rescoped to that target.
|
||||
|
||||
:param session: a session object to send out HTTP requests.
|
||||
:type session: keystoneclient.session.Session
|
||||
|
||||
:returns: a token data representation
|
||||
:rtype: :py:class:`keystoneclient.access.AccessInfo`
|
||||
|
||||
"""
|
||||
auth_ref = self.get_unscoped_auth_ref(session)
|
||||
scoping = self._get_scoping_data()
|
||||
|
||||
if any(scoping.values()):
|
||||
token_plugin = self.rescoping_plugin(self.auth_url,
|
||||
token=auth_ref.auth_token,
|
||||
**scoping)
|
||||
|
||||
auth_ref = token_plugin.get_auth_ref(session)
|
||||
|
||||
return auth_ref
|
||||
|
||||
@abc.abstractmethod
|
||||
def get_unscoped_auth_ref(self, session, **kwargs):
|
||||
"""Fetch unscoped federated token."""
|
||||
|
|
@ -0,0 +1,96 @@
|
|||
# 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 uuid
|
||||
|
||||
from keystoneclient import access
|
||||
from keystoneclient.auth.identity import v3
|
||||
from keystoneclient import fixture
|
||||
from keystoneclient import session
|
||||
from keystoneclient.tests.unit import utils
|
||||
|
||||
|
||||
class TesterFederationPlugin(v3.FederatedBaseAuth):
|
||||
|
||||
def get_unscoped_auth_ref(self, sess, **kwargs):
|
||||
# This would go and talk to an idp or something
|
||||
resp = sess.post(self.federated_token_url, authenticated=False)
|
||||
return access.AccessInfo.factory(resp=resp, body=resp.json())
|
||||
|
||||
|
||||
class V3FederatedPlugin(utils.TestCase):
|
||||
|
||||
AUTH_URL = 'http://keystone/v3'
|
||||
|
||||
def setUp(self):
|
||||
super(V3FederatedPlugin, self).setUp()
|
||||
|
||||
self.unscoped_token = fixture.V3Token()
|
||||
self.unscoped_token_id = uuid.uuid4().hex
|
||||
self.scoped_token = copy.deepcopy(self.unscoped_token)
|
||||
self.scoped_token.set_project_scope()
|
||||
self.scoped_token.methods.append('token')
|
||||
self.scoped_token_id = uuid.uuid4().hex
|
||||
|
||||
s = self.scoped_token.add_service('compute', name='nova')
|
||||
s.add_standard_endpoints(public='http://nova/public',
|
||||
admin='http://nova/admin',
|
||||
internal='http://nova/internal')
|
||||
|
||||
self.idp = uuid.uuid4().hex
|
||||
self.protocol = uuid.uuid4().hex
|
||||
|
||||
self.token_url = ('%s/OS-FEDERATION/identity_providers/%s/protocols/%s'
|
||||
'/auth' % (self.AUTH_URL, self.idp, self.protocol))
|
||||
|
||||
headers = {'X-Subject-Token': self.unscoped_token_id}
|
||||
self.unscoped_mock = self.requests_mock.post(self.token_url,
|
||||
json=self.unscoped_token,
|
||||
headers=headers)
|
||||
|
||||
headers = {'X-Subject-Token': self.scoped_token_id}
|
||||
auth_url = self.AUTH_URL + '/auth/tokens'
|
||||
self.scoped_mock = self.requests_mock.post(auth_url,
|
||||
json=self.scoped_token,
|
||||
headers=headers)
|
||||
|
||||
def get_plugin(self, **kwargs):
|
||||
kwargs.setdefault('auth_url', self.AUTH_URL)
|
||||
kwargs.setdefault('protocol', self.protocol)
|
||||
kwargs.setdefault('identity_provider', self.idp)
|
||||
return TesterFederationPlugin(**kwargs)
|
||||
|
||||
def test_federated_url(self):
|
||||
plugin = self.get_plugin()
|
||||
self.assertEqual(self.token_url, plugin.federated_token_url)
|
||||
|
||||
def test_unscoped_behaviour(self):
|
||||
sess = session.Session(auth=self.get_plugin())
|
||||
self.assertEqual(self.unscoped_token_id, sess.get_token())
|
||||
|
||||
self.assertTrue(self.unscoped_mock.called)
|
||||
self.assertFalse(self.scoped_mock.called)
|
||||
|
||||
def test_scoped_behaviour(self):
|
||||
auth = self.get_plugin(project_id=self.scoped_token.project_id)
|
||||
sess = session.Session(auth=auth)
|
||||
self.assertEqual(self.scoped_token_id, sess.get_token())
|
||||
|
||||
self.assertTrue(self.unscoped_mock.called)
|
||||
self.assertTrue(self.scoped_mock.called)
|
||||
|
||||
def test_options(self):
|
||||
opts = [o.name for o in v3.FederatedBaseAuth.get_options()]
|
||||
|
||||
self.assertIn('protocol', opts)
|
||||
self.assertIn('identity-provider', opts)
|
||||
Loading…
Reference in New Issue