openstack
/
django_openstack_auth
Code Issues Proposed changes

Merge "Use keystone auth plugins"

This commit is contained in:
Jenkins 2015-01-28 23:57:49 +00:00 committed by Gerrit Code Review
parent cf129a6883 01e0abc17d
commit b7133dc77a
5 changed files with 281 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
openstack_auth/backend.py
View File

@ -69,35 +69,22 @@ class KeystoneBackend(object):
"""Authenticates a user via the Keystone Identity API."""
LOG.debug('Beginning user authentication for user "%s".' % username)
insecure = getattr(settings, 'OPENSTACK_SSL_NO_VERIFY', False)
ca_cert = getattr(settings, "OPENSTACK_SSL_CACERT", None)
endpoint_type = getattr(
settings, 'OPENSTACK_ENDPOINT_TYPE', 'publicURL')
interface = getattr(settings, 'OPENSTACK_ENDPOINT_TYPE', 'public')
if auth_url is None:
auth_url = settings.OPENSTACK_KEYSTONE_URL
# keystone client v3 does not support logging in on the v2 url any more
if utils.get_keystone_version() >= 3:
if utils.has_in_url_path(auth_url, "/v2.0"):
LOG.warning("The settings.py file points to a v2.0 keystone "
"endpoint, but v3 is specified as the API version "
"to use. Using v3 endpoint for authentication.")
auth_url = utils.url_path_replace(auth_url, "/v2.0", "/v3", 1)
session = utils.get_session()
keystone_client_class = utils.get_keystone_client().Client
auth_url = utils.fix_auth_url_version(auth_url)
unscoped_auth = utils.get_password_auth_plugin(auth_url,
username,
password,
user_domain_name)
keystone_client = utils.get_keystone_client()
try:
client = keystone_client.Client(
user_domain_name=user_domain_name,
username=username,
password=password,
auth_url=auth_url,
insecure=insecure,
cacert=ca_cert,
debug=settings.DEBUG)
unscoped_auth_ref = client.auth_ref
unscoped_token = auth_user.Token(auth_ref=unscoped_auth_ref)
unscoped_auth_ref = unscoped_auth.get_access(session)
except (keystone_exceptions.Unauthorized,
keystone_exceptions.Forbidden,
keystone_exceptions.NotFound) as exc:
@ -114,22 +101,16 @@ class KeystoneBackend(object):
# Check expiry for our unscoped auth ref.
self.check_auth_expiry(unscoped_auth_ref)
# Check if token is automatically scoped to default_project
# grab the project from this token, to use as a default
# if no recent_project is found in the cookie
token_proj_id = None
if unscoped_auth_ref.project_scoped:
token_proj_id = unscoped_auth_ref.get('project',
{}).get('id')
unscoped_client = keystone_client_class(session=session,
auth=unscoped_auth)
# We list all the user's projects
try:
if utils.get_keystone_version() < 3:
projects = client.tenants.list()
else:
client.management_url = auth_url
projects = client.projects.list(
if utils.get_keystone_version() >= 3:
projects = unscoped_client.projects.list(
user=unscoped_auth_ref.user_id)
else:
projects = unscoped_client.tenants.list()
except (keystone_exceptions.ClientException,
keystone_exceptions.AuthorizationFailure) as exc:
msg = _('Unable to retrieve authorized projects.')
@ -143,51 +124,55 @@ class KeystoneBackend(object):
# the recent project id a user might have set in a cookie
recent_project = None
if request:
# Check if token is automatically scoped to default_project
# grab the project from this token, to use as a default
# if no recent_project is found in the cookie
recent_project = request.COOKIES.get('recent_project',
token_proj_id)
unscoped_auth_ref.project_id)
# if a most recent project was found, try using it first
for pos, project in enumerate(projects):
if project.id == recent_project:
# move recent project to the beginning
projects.pop(pos)
projects.insert(0, project)
break
if recent_project:
for pos, project in enumerate(projects):
if project.id == recent_project:
# move recent project to the beginning
projects.pop(pos)
projects.insert(0, project)
break
for project in projects:
token = unscoped_auth_ref.auth_token
scoped_auth = utils.get_token_auth_plugin(auth_url,
token=token,
project_id=project.id)
try:
client = keystone_client.Client(
tenant_id=project.id,
token=unscoped_auth_ref.auth_token,
auth_url=auth_url,
insecure=insecure,
cacert=ca_cert,
debug=settings.DEBUG)
auth_ref = client.auth_ref
break
scoped_auth_ref = scoped_auth.get_access(session)
except (keystone_exceptions.ClientException,
keystone_exceptions.AuthorizationFailure):
auth_ref = None
if auth_ref is None:
pass
else:
break
else:
msg = _("Unable to authenticate to any available projects.")
raise exceptions.KeystoneAuthException(msg)
# Check expiry for our new scoped token.
self.check_auth_expiry(auth_ref)
self.check_auth_expiry(scoped_auth_ref)
# If we made it here we succeeded. Create our User!
user = auth_user.create_user_from_token(
request,
auth_user.Token(auth_ref),
client.service_catalog.url_for(endpoint_type=endpoint_type))
auth_user.Token(scoped_auth_ref),
scoped_auth_ref.service_catalog.url_for(endpoint_type=interface))
if request is not None:
request.session['unscoped_token'] = unscoped_token.id
request.session['unscoped_token'] = unscoped_auth_ref.auth_token
request.user = user
scoped_client = keystone_client_class(session=session,
auth=scoped_auth)
# Support client caching to save on auth calls.
setattr(request, KEYSTONE_CLIENT_ATTR, client)
setattr(request, KEYSTONE_CLIENT_ATTR, scoped_client)
LOG.debug('Authentication completed for user "%s".' % username)
return user

272
openstack_auth/tests/tests.py
View File

@ -16,7 +16,11 @@ from django.contrib import auth
from django.core.urlresolvers import reverse
from django import http
from django import test
from keystoneclient.auth.identity import v2 as auth_v2
from keystoneclient.auth.identity import v3 as auth_v3
from keystoneclient.auth import token_endpoint
from keystoneclient import exceptions as keystone_exceptions
from keystoneclient import session
from keystoneclient.v2_0 import client as client_v2
from keystoneclient.v3 import client as client_v3
import mock
@ -43,64 +47,44 @@ class OpenStackAuthTestsMixin(object):
('admin', {'interface': 'adminURL'})
]
def tearDown(self):
self.mox.UnsetStubs()
self.mox.VerifyAll()
def _mock_unscoped_client(self, user):
self.mox.StubOutWithMock(self.ks_client_module, "Client")
self.ks_client_module.Client(auth_url=settings.OPENSTACK_KEYSTONE_URL,
password=user.password,
username=user.name,
user_domain_name=DEFAULT_DOMAIN,
insecure=False,
cacert=None,
debug=False)\
.AndReturn(self.keystone_client_unscoped)
plugin = self._create_password_auth()
plugin.get_access(mox.IsA(session.Session)). \
AndReturn(self.data.unscoped_access_info)
return self.ks_client_module.Client(session=mox.IsA(session.Session),
auth=plugin)
def _mock_unscoped_client_with_token(self, user, unscoped):
self.mox.StubOutWithMock(self.ks_client_module, "Client")
url = settings.OPENSTACK_KEYSTONE_URL
self.ks_client_module.Client(user_id=user.id,
auth_url=url,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)\
.AndReturn(self.keystone_client_unscoped)
plugin = token_endpoint.Token(settings.OPENSTACK_KEYSTONE_URL,
unscoped.auth_token)
return self.ks_client_module.Client(session=mox.IsA(session.Session),
auth=plugin)
def _mock_client_token_auth_failure(self, unscoped, tenant_id):
exc = keystone_exceptions.AuthorizationFailure
self.ks_client_module.Client(auth_url=settings.OPENSTACK_KEYSTONE_URL,
tenant_id=tenant_id,
insecure=False,
cacert=None,
token=unscoped.auth_token,
debug=False) \
.AndRaise(exc)
plugin = self._create_token_auth(tenant_id, unscoped.auth_token)
plugin.get_access(mox.IsA(session.Session)). \
AndRaise(keystone_exceptions.AuthorizationFailure)
def _mock_client_password_auth_failure(self, username, password, exc):
self.mox.StubOutWithMock(self.ks_client_module, "Client")
self.ks_client_module.Client(auth_url=settings.OPENSTACK_KEYSTONE_URL,
password=password,
username=username,
user_domain_name=DEFAULT_DOMAIN,
insecure=False,
cacert=None,
debug=False).AndRaise(exc)
plugin = self._create_password_auth(username=username,
password=password)
plugin.get_access(mox.IsA(session.Session)).AndRaise(exc)
def _mock_scoped_client_for_tenant(self, auth_ref, tenant_id, url=None):
def _mock_scoped_client_for_tenant(self, auth_ref, tenant_id, url=None,
client=True):
if url is None:
auth_url = settings.OPENSTACK_KEYSTONE_URL
else:
auth_url = url
self.ks_client_module.Client(auth_url=auth_url,
tenant_id=tenant_id,
insecure=False,
cacert=None,
token=auth_ref.auth_token,
debug=False) \
.AndReturn(self.keystone_client_scoped)
url = settings.OPENSTACK_KEYSTONE_URL
plugin = self._create_token_auth(
tenant_id,
token=self.data.unscoped_access_info.auth_token,
url=url)
plugin.get_access(mox.IsA(session.Session)).AndReturn(auth_ref)
if client:
return self.ks_client_module.Client(
session=mox.IsA(session.Session),
auth=plugin)
def get_form_data(self, user):
return {'region': settings.OPENSTACK_KEYSTONE_URL,
@ -120,25 +104,67 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
self.addCleanup(override.disable)
self.mox = mox.Mox()
self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self.mox.UnsetStubs)
self.data = data_v2.generate_test_data()
self.ks_client_module = client_v2
endpoint = settings.OPENSTACK_KEYSTONE_URL
self.keystone_client_unscoped = self.ks_client_module.Client(
endpoint=endpoint,
auth_ref=self.data.unscoped_access_info)
self.keystone_client_scoped = self.ks_client_module.Client(
endpoint=endpoint,
auth_ref=self.data.scoped_access_info)
settings.OPENSTACK_API_VERSIONS['identity'] = 2.0
settings.OPENSTACK_KEYSTONE_URL = "http://localhost:5000/v2.0"
def _mock_unscoped_list_tenants(self, tenants):
self.mox.StubOutWithMock(self.keystone_client_unscoped.tenants, "list")
self.keystone_client_unscoped.tenants.list().AndReturn(tenants)
self.mox.StubOutClassWithMocks(token_endpoint, 'Token')
self.mox.StubOutClassWithMocks(auth_v2, 'Token')
self.mox.StubOutClassWithMocks(auth_v2, 'Password')
self.mox.StubOutClassWithMocks(client_v2, 'Client')
def _mock_unscoped_list_tenants(self, client, tenants):
client.tenants = self.mox.CreateMockAnything()
client.tenants.list().AndReturn(tenants)
def _mock_unscoped_client_list_tenants(self, user, tenants):
self._mock_unscoped_client(user)
self._mock_unscoped_list_tenants(tenants)
client = self._mock_unscoped_client(user)
self._mock_unscoped_list_tenants(client, tenants)
def _mock_client_delete_token(self, user, token, url=None):
if not url:
url = settings.OPENSTACK_KEYSTONE_URL
plugin = token_endpoint.Token(
endpoint=url,
token=self.data.unscoped_access_info.auth_token)
client = self.ks_client_module.Client(session=mox.IsA(session.Session),
auth=plugin)
client.tokens = self.mox.CreateMockAnything()
client.tokens.delete(token=token)
return client
def _create_password_auth(self, username=None, password=None, url=None):
if not username:
username = self.data.user.name
if not password:
password = self.data.user.password
if not url:
url = settings.OPENSTACK_KEYSTONE_URL
return auth_v2.Password(auth_url=url,
password=password,
username=username)
def _create_token_auth(self, project_id, token=None, url=None):
if not token:
token = self.data.unscoped_access_info.auth_token
if not url:
url = settings.OPENSTACK_KEYSTONE_URL
return auth_v2.Token(auth_url=url,
token=token,
tenant_id=project_id,
reauthenticate=False)
def _login(self):
tenants = [self.data.tenant_one, self.data.tenant_two]
@ -309,13 +335,16 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
scoped = self.data.scoped_access_info
sc = self.data.service_catalog
et = getattr(settings, 'OPENSTACK_ENDPOINT_TYPE', 'publicURL')
endpoint = sc.url_for(endpoint_type=et)
form_data = self.get_form_data(user)
self._mock_unscoped_client_list_tenants(user, tenants)
self._mock_scoped_client_for_tenant(unscoped, self.data.tenant_one.id)
self._mock_scoped_client_for_tenant(scoped, tenant.id,
url=sc.url_for(endpoint_type=et))
self._mock_client_delete_token(user, unscoped.auth_token, endpoint)
self._mock_scoped_client_for_tenant(scoped, tenant.id, url=endpoint,
client=False)
self.mox.ReplayAll()
url = reverse('login')
@ -350,13 +379,13 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
def test_switch_region(self, next=None):
tenants = [self.data.tenant_one, self.data.tenant_two]
user = self.data.user
unscoped = self.data.unscoped_access_info
scoped = self.data.scoped_access_info
sc = self.data.service_catalog
form_data = self.get_form_data(user)
self._mock_unscoped_client_list_tenants(user, tenants)
self._mock_scoped_client_for_tenant(unscoped, self.data.tenant_one.id)
self._mock_scoped_client_for_tenant(scoped, self.data.tenant_one.id)
self.mox.ReplayAll()
@ -399,18 +428,15 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
user = self.data.user
unscoped = self.data.unscoped_access_info
self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_tenants(tenants)
client = self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_tenants(client, tenants)
self.mox.ReplayAll()
tenant_list = utils.get_project_list(
user_id=user.id,
auth_url=settings.OPENSTACK_KEYSTONE_URL,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)
token=unscoped.auth_token)
self.assertEqual(tenant_list, expected_tenants)
def test_tenant_list_caching(self):
@ -419,17 +445,14 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
user = self.data.user
unscoped = self.data.unscoped_access_info
self._mock_unscoped_list_tenants(tenants)
self._mock_unscoped_client_with_token(user, unscoped)
client = self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_tenants(client, tenants)
self.mox.ReplayAll()
tenant_list = utils.get_project_list(
user_id=user.id,
auth_url=settings.OPENSTACK_KEYSTONE_URL,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)
token=unscoped.auth_token)
self.assertEqual(tenant_list, expected_tenants)
# Test to validate that requesting the project list again results
@ -439,10 +462,7 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
tenant_list = utils.get_project_list(
user_id=user.id,
auth_url=settings.OPENSTACK_KEYSTONE_URL,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)
token=unscoped.auth_token)
self.assertEqual(tenant_list, expected_tenants)
utils.remove_project_cache(unscoped.auth_token)
@ -452,14 +472,39 @@ class OpenStackAuthTestsV2(OpenStackAuthTestsMixin, test.TestCase):
class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
def _mock_unscoped_client_list_projects(self, user, projects):
self._mock_unscoped_client(user)
self._mock_unscoped_list_projects(user, projects)
client = self._mock_unscoped_client(user)
self._mock_unscoped_list_projects(client, user, projects)
def _mock_unscoped_list_projects(self, user, projects):
self.mox.StubOutWithMock(self.keystone_client_unscoped.projects,
"list")
self.keystone_client_unscoped.projects.list(user=user.id) \
.AndReturn(projects)
def _mock_unscoped_list_projects(self, client, user, projects):
client.projects = self.mox.CreateMockAnything()
client.projects.list(user=user.id).AndReturn(projects)
def _create_password_auth(self, username=None, password=None, url=None):
if not username:
username = self.data.user.name
if not password:
password = self.data.user.password
if not url:
url = settings.OPENSTACK_KEYSTONE_URL
return auth_v3.Password(auth_url=url,
password=password,
username=username,
user_domain_name=DEFAULT_DOMAIN)
def _create_token_auth(self, project_id, token=None, url=None):
if not token:
token = self.data.unscoped_access_info.auth_token
if not url:
url = settings.OPENSTACK_KEYSTONE_URL
return auth_v3.Token(auth_url=url,
token=token,
project_id=project_id,
reauthenticate=False)
def setUp(self):
super(OpenStackAuthTestsV3, self).setUp()
@ -470,18 +515,20 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
self.addCleanup(override.disable)
self.mox = mox.Mox()
self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self.mox.UnsetStubs)
self.data = data_v3.generate_test_data()
self.ks_client_module = client_v3
endpoint = settings.OPENSTACK_KEYSTONE_URL
self.keystone_client_unscoped = self.ks_client_module.Client(
endpoint=endpoint,
auth_ref=self.data.unscoped_access_info)
self.keystone_client_scoped = self.ks_client_module.Client(
endpoint=endpoint,
auth_ref=self.data.scoped_access_info)
settings.OPENSTACK_API_VERSIONS['identity'] = 3
settings.OPENSTACK_KEYSTONE_URL = "http://localhost:5000/v3"
self.mox.StubOutClassWithMocks(token_endpoint, 'Token')
self.mox.StubOutClassWithMocks(auth_v3, 'Token')
self.mox.StubOutClassWithMocks(auth_v3, 'Password')
self.mox.StubOutClassWithMocks(client_v3, 'Client')
def test_login(self):
projects = [self.data.project_one, self.data.project_two]
user = self.data.user
@ -619,7 +666,6 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
project = self.data.project_two
projects = [self.data.project_one, self.data.project_two]
user = self.data.user
unscoped = self.data.unscoped_access_info
scoped = self.data.scoped_access_info
sc = self.data.service_catalog
et = getattr(settings, 'OPENSTACK_ENDPOINT_TYPE', 'publicURL')
@ -627,11 +673,12 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
form_data = self.get_form_data(user)
self._mock_unscoped_client_list_projects(user, projects)
self._mock_scoped_client_for_tenant(unscoped, self.data.project_one.id)
self._mock_scoped_client_for_tenant(scoped, self.data.project_one.id)
self._mock_scoped_client_for_tenant(
unscoped,
scoped,
project.id,
url=sc.url_for(endpoint_type=et))
url=sc.url_for(endpoint_type=et),
client=False)
self.mox.ReplayAll()
@ -667,12 +714,12 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
def test_switch_region(self, next=None):
projects = [self.data.project_one, self.data.project_two]
user = self.data.user
unscoped = self.data.unscoped_access_info
scoped = self.data.unscoped_access_info
sc = self.data.service_catalog
form_data = self.get_form_data(user)
self._mock_unscoped_client_list_projects(user, projects)
self._mock_scoped_client_for_tenant(unscoped, self.data.project_one.id)
self._mock_scoped_client_for_tenant(scoped, self.data.project_one.id)
self.mox.ReplayAll()
@ -714,17 +761,14 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
user = self.data.user
unscoped = self.data.unscoped_access_info
self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_projects(user, projects)
client = self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_projects(client, user, projects)
self.mox.ReplayAll()
project_list = utils.get_project_list(
user_id=user.id,
auth_url=settings.OPENSTACK_KEYSTONE_URL,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)
token=unscoped.auth_token)
self.assertEqual(project_list, expected_projects)
def test_tenant_list_caching(self):
@ -733,18 +777,15 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
user = self.data.user
unscoped = self.data.unscoped_access_info
self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_projects(user, projects)
client = self._mock_unscoped_client_with_token(user, unscoped)
self._mock_unscoped_list_projects(client, user, projects)
self.mox.ReplayAll()
project_list = utils.get_project_list(
user_id=user.id,
auth_url=settings.OPENSTACK_KEYSTONE_URL,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)
token=unscoped.auth_token)
self.assertEqual(project_list, expected_projects)
# Test to validate that requesting the project list again results
@ -754,10 +795,7 @@ class OpenStackAuthTestsV3(OpenStackAuthTestsMixin, test.TestCase):
project_list = utils.get_project_list(
user_id=user.id,
auth_url=settings.OPENSTACK_KEYSTONE_URL,
token=unscoped.auth_token,
insecure=False,
cacert=None,
debug=False)
token=unscoped.auth_token)
self.assertEqual(project_list, expected_projects)
utils.remove_project_cache(unscoped.auth_token)

8
openstack_auth/user.py
View File

@ -273,9 +273,6 @@ class User(models.AnonymousUser):
@property
def authorized_tenants(self):
"""Returns a memoized list of tenants this user may access."""
insecure = getattr(settings, 'OPENSTACK_SSL_NO_VERIFY', False)
ca_cert = getattr(settings, "OPENSTACK_SSL_CACERT", None)
if self.is_authenticated() and self._authorized_tenants is None:
endpoint = self.endpoint
token = self.token
@ -283,10 +280,7 @@ class User(models.AnonymousUser):
self._authorized_tenants = utils.get_project_list(
user_id=self.id,
auth_url=endpoint,
token=token.id,
insecure=insecure,
cacert=ca_cert,
debug=settings.DEBUG)
token=token.id)
except (keystone_exceptions.ClientException,
keystone_exceptions.AuthorizationFailure):
LOG.exception('Unable to retrieve project list.')

72
openstack_auth/utils.py
View File

@ -21,6 +21,10 @@ from django.contrib.auth import middleware
from django.contrib.auth import models
from django.utils import decorators
from django.utils import timezone
from keystoneclient.auth.identity import v2 as v2_auth
from keystoneclient.auth.identity import v3 as v3_auth
from keystoneclient.auth import token_endpoint
from keystoneclient import session
from keystoneclient.v2_0 import client as client_v2
from keystoneclient.v3 import client as client_v3
from six.moves.urllib import parse as urlparse
@ -152,6 +156,16 @@ def get_keystone_version():
return getattr(settings, 'OPENSTACK_API_VERSIONS', {}).get('identity', 2.0)
def get_session():
insecure = getattr(settings, 'OPENSTACK_SSL_NO_VERIFY', False)
verify = getattr(settings, 'OPENSTACK_SSL_CACERT', True)
if insecure:
verify = False
return session.Session(verify=verify)
def get_keystone_client():
if get_keystone_version() < 3:
return client_v2
@ -180,20 +194,60 @@ def url_path_replace(url, old, new, count=None):
scheme, netloc, path.replace(old, new, *args), query, fragment))
def fix_auth_url_version(auth_url):
"""Fix up the auth url if an invalid version prefix was given.
People still give a v2 auth_url even when they specify that they want v3
authentication. Fix the URL to say v3. This should be smarter and take the
base, unversioned URL and discovery.
"""
if get_keystone_version() >= 3:
if has_in_url_path(auth_url, "/v2.0"):
LOG.warning("The settings.py file points to a v2.0 keystone "
"endpoint, but v3 is specified as the API version "
"to use. Using v3 endpoint for authentication.")
auth_url = url_path_replace(auth_url, "/v2.0", "/v3", 1)
return auth_url
def get_password_auth_plugin(auth_url, username, password, user_domain_name):
if get_keystone_version() >= 3:
return v3_auth.Password(auth_url=auth_url,
username=username,
password=password,
user_domain_name=user_domain_name)
else:
return v2_auth.Password(auth_url=auth_url,
username=username,
password=password)
def get_token_auth_plugin(auth_url, token, project_id):
if get_keystone_version() >= 3:
return v3_auth.Token(auth_url=auth_url,
token=token,
project_id=project_id,
reauthenticate=False)
else:
return v2_auth.Token(auth_url=auth_url,
token=token,
tenant_id=project_id,
reauthenticate=False)
@memoize_by_keyword_arg(_PROJECT_CACHE, ('token', ))
def get_project_list(*args, **kwargs):
sess = kwargs.get('session') or get_session()
auth_url = fix_auth_url_version(kwargs['auth_url'])
auth = token_endpoint.Token(auth_url, kwargs['token'])
client = get_keystone_client().Client(session=sess, auth=auth)
if get_keystone_version() < 3:
auth_url = url_path_replace(
kwargs.get('auth_url', ''), '/v3', '/v2.0', 1)
kwargs['auth_url'] = auth_url
client = get_keystone_client().Client(*args, **kwargs)
projects = client.tenants.list()
else:
auth_url = url_path_replace(
kwargs.get('auth_url', ''), '/v2.0', '/v3', 1)
kwargs['auth_url'] = auth_url
client = get_keystone_client().Client(*args, **kwargs)
client.management_url = auth_url
projects = client.projects.list(user=kwargs.get('user_id'))
projects.sort(key=lambda project: project.name.lower())

43
openstack_auth/views.py
View File

@ -24,6 +24,7 @@ from django.utils import http
from django.views.decorators.cache import never_cache # noqa
from django.views.decorators.csrf import csrf_protect # noqa
from django.views.decorators.debug import sensitive_post_parameters # noqa
from keystoneclient.auth import token_endpoint
from keystoneclient import exceptions as keystone_exceptions
from openstack_auth import forms
@ -136,23 +137,21 @@ def logout(request, login_url=None, **kwargs):
def delete_token(endpoint, token_id):
"""Delete a token."""
insecure = getattr(settings, 'OPENSTACK_SSL_NO_VERIFY', False)
ca_cert = getattr(settings, "OPENSTACK_SSL_CACERT", None)
utils.remove_project_cache(token_id)
try:
if utils.get_keystone_version() >= 3:
if not utils.has_in_url_path(endpoint, '/v3'):
endpoint = utils.url_path_replace(endpoint, '/v2.0', '/v3', 1)
client = utils.get_keystone_client().Client(
endpoint=endpoint,
token=token_id,
insecure=insecure,
cacert=ca_cert,
debug=settings.DEBUG)
endpoint = utils.fix_auth_url_version(endpoint)
session = utils.get_session()
auth_plugin = token_endpoint.Token(endpoint=endpoint,
token=token_id)
client = utils.get_keystone_client().Client(session=session,
auth=auth_plugin)
if utils.get_keystone_version() >= 3:
client.tokens.revoke_token(token=token_id)
else:
client.tokens.delete(token=token_id)
LOG.info('Deleted token %s' % token_id)
except keystone_exceptions.ClientException:
LOG.info('Could not delete token')
@ -163,21 +162,15 @@ def switch(request, tenant_id, redirect_field_name=auth.REDIRECT_FIELD_NAME):
"""Switches an authenticated user from one project to another."""
LOG.debug('Switching to tenant %s for user "%s".'
% (tenant_id, request.user.username))
insecure = getattr(settings, 'OPENSTACK_SSL_NO_VERIFY', False)
ca_cert = getattr(settings, "OPENSTACK_SSL_CACERT", None)
endpoint = request.user.endpoint
endpoint = utils.fix_auth_url_version(request.user.endpoint)
session = utils.get_session()
auth = utils.get_token_auth_plugin(auth_url=endpoint,
token=request.user.token.id,
project_id=tenant_id)
try:
if utils.get_keystone_version() >= 3:
if not utils.has_in_url_path(endpoint, '/v3'):
endpoint = utils.url_path_replace(endpoint, '/v2.0', '/v3', 1)
client = utils.get_keystone_client().Client(
tenant_id=tenant_id,
token=request.user.token.id,
auth_url=endpoint,
insecure=insecure,
cacert=ca_cert,
debug=settings.DEBUG)
auth_ref = client.auth_ref
auth_ref = auth.get_access(session)
msg = 'Project switch successful for user "%(username)s".' % \
{'username': request.user.username}
LOG.info(msg)