openstack
/
keystone
Code Issues Proposed changes

LDAP Identity backend 

Bug 933852

Merged over the code from the legacy keystone implementation, updated
style and streamlined the API a bit.

 * Unit tests can be run against a live OpenLDAP server
 * Password hashing done via passlib. Only does salted sha1, which is what simple_bind requires, but is not secure.
 * Added pip dependencies

Change-Id: I5296d94f6b7d0a7c7dbc887cdae872171e34bb5f
This commit is contained in:
Adam Young 2012-02-06 21:21:46 -05:00 committed by termie
parent 679fd363d8
commit 63437e9dca
21 changed files with 1718 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
AUTHORS
View File

@ -1,4 +1,5 @@
Adam Gandelman <adamg@canonical.com>
Adam Young <ayoung@redhat.com>
Adipudi Praveena <padipudi@padipudi.(none)>
Akira YOSHIYAMA <akirayoshiyama@gmail.com>
Alan Pevec <apevec@redhat.com>

10
etc/keystone.conf
View File

@ -22,6 +22,16 @@ min_pool_size = 5
max_pool_size = 10
pool_timeout = 200
[ldap]
#url = ldap://localhost
#tree_dn = dc=example,dc=com
#user_tree_dn = ou=Users,dc=example,dc=com
#role_tree_dn = ou=Roles,dc=example,dc=com
#tenant_tree_dn = ou=Groups,dc=example,dc=com
#user = dc=Manager,dc=example,dc=com
#password = freeipa4all
#suffix = cn=example,cn=com
[identity]
driver = keystone.identity.backends.kvs.Identity

1
keystone/common/ldap/__init__.py Normal file
View File

@ -0,0 +1 @@
from keystone.common.ldap.core import *

317
keystone/common/ldap/core.py Normal file
View File

@ -0,0 +1,317 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
import logging
import ldap
from keystone.common.ldap import fakeldap
LOG = logging.getLogger('keystone.common.ldap')
LDAP_VALUES = {'TRUE': True, 'FALSE': False}
def py2ldap(val):
if isinstance(val, str):
return val
elif isinstance(val, bool):
return 'TRUE' if val else 'FALSE'
else:
return str(val)
def ldap2py(val):
try:
return LDAP_VALUES[val]
except KeyError:
pass
try:
return int(val)
except ValueError:
pass
return val
def safe_iter(attrs):
if attrs is None:
return
elif isinstance(attrs, list):
for e in attrs:
yield e
else:
yield attrs
class BaseLdap(object):
DEFAULT_SUFFIX = "dc=example,dc=com"
DEFAULT_OU = None
DEFAULT_STRUCTURAL_CLASSES = None
DEFAULT_ID_ATTR = 'cn'
DEFAULT_OBJECTCLASS = None
DUMB_MEMBER_DN = 'cn=dumb,dc=nonexistent'
options_name = None
model = None
attribute_mapping = {}
attribute_ignore = []
model = None
tree_dn = None
def __init__(self, conf):
self.LDAP_URL = conf.ldap.url
self.LDAP_USER = conf.ldap.user
self.LDAP_PASSWORD = conf.ldap.password
if self.options_name is not None:
self.suffix = conf.ldap.suffix
if (self.suffix == None):
self.suffix = self.DEFAULT_SUFFIX
dn = '%s_tree_dn' % self.options_name
self.tree_dn = (getattr(conf.ldap, dn)
or '%s,%s' % (self.suffix, self.DEFAULT_OU))
idatt = '%s_id_attribute' % self.options_name
self.id_attr = getattr(conf.ldap, idatt) or self.DEFAULT_ID_ATTR
objclass = '%s_objectclass' % self.options_name
self.object_class = (getattr(conf.ldap, objclass)
or self.DEFAULT_OBJECTCLASS)
self.structural_classes = self.DEFAULT_STRUCTURAL_CLASSES
self.use_dumb_member = conf.ldap.use_dumb_member
def get_connection(self, user=None, password=None):
if self.LDAP_URL.startswith('fake://'):
conn = fakeldap.FakeLdap(self.LDAP_URL)
else:
conn = LdapWrapper(self.LDAP_URL)
if user is None:
user = self.LDAP_USER
if password is None:
password = self.LDAP_PASSWORD
conn.simple_bind_s(user, password)
return conn
def _id_to_dn(self, id):
return '%s=%s,%s' % (self.id_attr,
ldap.dn.escape_dn_chars(str(id)),
self.tree_dn)
@staticmethod
def _dn_to_id(dn):
return ldap.dn.str2dn(dn)[0][0][1]
def _ldap_res_to_model(self, res):
obj = self.model(id=self._dn_to_id(res[0]))
for k in obj.known_keys:
if k in self.attribute_ignore:
continue
try:
v = res[1][self.attribute_mapping.get(k, k)]
except KeyError:
pass
else:
try:
obj[k] = v[0]
except IndexError:
obj[k] = None
return obj
def affirm_unique(self, values):
if values['name'] is not None:
entity = self.get_by_name(values['name'])
if entity is not None:
raise Exception('%s with id %s already exists'
% (self.options_name, values['id']))
if values['id'] is not None:
entity = self.get(values['id'])
if entity is not None:
raise Exception('%s with id %s already exists'
% (self.options_name, values['id']))
def create(self, values):
conn = self.get_connection()
object_classes = self.structural_classes + [self.object_class]
attrs = [('objectClass', object_classes)]
for k, v in values.iteritems():
if k == 'id' or k in self.attribute_ignore:
continue
if v is not None:
attr_type = self.attribute_mapping.get(k, k)
attrs.append((attr_type, [v]))
if 'groupOfNames' in object_classes and self.use_dumb_member:
attrs.append(('member', [self.DUMB_MEMBER_DN]))
conn.add_s(self._id_to_dn(values['id']), attrs)
return values
def _ldap_get(self, id, filter=None):
conn = self.get_connection()
query = '(objectClass=%s)' % self.object_class
if filter is not None:
query = '(&%s%s)' % (filter, query)
try:
res = conn.search_s(self._id_to_dn(id), ldap.SCOPE_BASE, query)
except ldap.NO_SUCH_OBJECT:
return None
try:
return res[0]
except IndexError:
return None
def _ldap_get_all(self, filter=None):
conn = self.get_connection()
query = '(objectClass=%s)' % (self.object_class,)
if filter is not None:
query = '(&%s%s)' % (filter, query)
try:
return conn.search_s(self.tree_dn, ldap.SCOPE_ONELEVEL, query)
except ldap.NO_SUCH_OBJECT:
return []
def get(self, id, filter=None):
res = self._ldap_get(id, filter)
if res is None:
return None
else:
return self._ldap_res_to_model(res)
def get_all(self, filter=None):
return [self._ldap_res_to_model(x)
for x in self._ldap_get_all(filter)]
def get_page(self, marker, limit):
return self._get_page(marker, limit, self.get_all())
def get_page_markers(self, marker, limit):
return self._get_page_markers(marker, limit, self.get_all())
@staticmethod
def _get_page(marker, limit, lst, key=lambda x: x.id):
lst.sort(key=key)
if not marker:
return lst[:limit]
else:
return [x for x in lst if key(x) > marker][:limit]
@staticmethod
def _get_page_markers(marker, limit, lst, key=lambda x: x.id):
if len(lst) < limit:
return (None, None)
lst.sort(key=key)
if marker is None:
if len(lst) <= limit + 1:
nxt = None
else:
nxt = key(lst[limit])
return (None, nxt)
i = 0
for i, item in enumerate(lst):
k = key(item)
if k >= marker:
break
if i <= limit:
prv = None
else:
prv = key(lst[i - limit])
if i + limit >= len(lst) - 1:
nxt = None
else:
nxt = key(lst[i + limit])
return (prv, nxt)
def update(self, id, values, old_obj=None):
if old_obj is None:
old_obj = self.get(id)
modlist = []
for k, v in values.iteritems():
if k == 'id' or k in self.attribute_ignore:
continue
if v is None:
if old_obj[k] is not None:
modlist.append((ldap.MOD_DELETE,
self.attribute_mapping.get(k, k),
None))
elif old_obj[k] != v:
if old_obj[k] is None:
op = ldap.MOD_ADD
else:
op = ldap.MOD_REPLACE
modlist.append((op, self.attribute_mapping.get(k, k), [v]))
conn = self.get_connection()
conn.modify_s(self._id_to_dn(id), modlist)
def delete(self, id):
conn = self.get_connection()
conn.delete_s(self._id_to_dn(id))
class LdapWrapper(object):
def __init__(self, url):
LOG.debug("LDAP init: url=%s", url)
self.conn = ldap.initialize(url)
def simple_bind_s(self, user, password):
LOG.debug("LDAP bind: dn=%s", user)
return self.conn.simple_bind_s(user, password)
def add_s(self, dn, attrs):
ldap_attrs = [(kind, [py2ldap(x) for x in safe_iter(values)])
for kind, values in attrs]
if LOG.isEnabledFor(logging.DEBUG):
sane_attrs = [(kind, values
if kind != 'userPassword'
else ['****'])
for kind, values in ldap_attrs]
LOG.debug('LDAP add: dn=%s, attrs=%s', dn, sane_attrs)
return self.conn.add_s(dn, ldap_attrs)
def search_s(self, dn, scope, query):
if LOG.isEnabledFor(logging.DEBUG):
LOG.debug('LDAP search: dn=%s, scope=%s, query=%s',
dn,
fakeldap.scope_names[scope],
query)
res = self.conn.search_s(dn, scope, query)
o = []
for dn, attrs in res:
o.append((dn, dict((kind, [ldap2py(x) for x in values])
for kind, values in attrs.iteritems())))
return o
def modify_s(self, dn, modlist):
ldap_modlist = [
(op, kind, (None if values is None
else [py2ldap(x) for x in safe_iter(values)]))
for op, kind, values in modlist]
if LOG.isEnabledFor(logging.DEBUG):
sane_modlist = [(op, kind, (values if kind != 'userPassword'
else ['****']))
for op, kind, values in ldap_modlist]
LOG.debug("LDAP modify: dn=%s, modlist=%s", dn, sane_modlist)
return self.conn.modify_s(dn, ldap_modlist)
def delete_s(self, dn):
LOG.debug("LDAP delete: dn=%s", dn)
return self.conn.delete_s(dn)

313
keystone/common/ldap/fakeldap.py Normal file
View File

@ -0,0 +1,313 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# 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.
"""Fake LDAP server for test harness.
This class does very little error checking, and knows nothing about ldap
class definitions. It implements the minimum emulation of the python ldap
library to work with nova.
"""
import logging
import re
import shelve
import ldap
from keystone.common import utils
SCOPE_NAMES = {
ldap.SCOPE_BASE: 'SCOPE_BASE',
ldap.SCOPE_ONELEVEL: 'SCOPE_ONELEVEL',
ldap.SCOPE_SUBTREE: 'SCOPE_SUBTREE',
}
LOG = logging.getLogger('keystone.backends.ldap.fakeldap')
def _match_query(query, attrs):
"""Match an ldap query to an attribute dictionary.
The characters &, |, and ! are supported in the query. No syntax checking
is performed, so malformed querys will not work correctly.
"""
# cut off the parentheses
inner = query[1:-1]
if inner.startswith('&'):
# cut off the &
l, r = _paren_groups(inner[1:])
return _match_query(l, attrs) and _match_query(r, attrs)
if inner.startswith('|'):
# cut off the |
l, r = _paren_groups(inner[1:])
return _match_query(l, attrs) or _match_query(r, attrs)
if inner.startswith('!'):
# cut off the ! and the nested parentheses
return not _match_query(query[2:-1], attrs)
(k, _sep, v) = inner.partition('=')
return _match(k, v, attrs)
def _paren_groups(source):
"""Split a string into parenthesized groups."""
count = 0
start = 0
result = []
for pos in xrange(len(source)):
if source[pos] == '(':
if count == 0:
start = pos
count += 1
if source[pos] == ')':
count -= 1
if count == 0:
result.append(source[start:pos + 1])
return result
def _match(key, value, attrs):
"""Match a given key and value against an attribute list."""
if key not in attrs:
return False
# This is a wild card search. Implemented as all or nothing for now.
if value == '*':
return True
if key == 'serviceId':
# for serviceId, the backend is returning a list of numbers
# make sure we convert them to strings first before comparing
# them
str_sids = [str(x) for x in attrs[key]]
return str(value) in str_sids
if key != 'objectclass':
return value in attrs[key]
# it is an objectclass check, so check subclasses
values = _subs(value)
for v in values:
if v in attrs[key]:
return True
return False
def _subs(value):
"""Returns a list of subclass strings.
The strings represent the ldap objectclass plus any subclasses that
inherit from it. Fakeldap doesn't know about the ldap object structure,
so subclasses need to be defined manually in the dictionary below.
"""
subs = {'groupOfNames': ['keystoneTenant',
'keystoneRole',
'keystoneTenantRole']}
if value in subs:
return [value] + subs[value]
return [value]
server_fail = False
class FakeShelve(dict):
@classmethod
def get_instance(cls):
try:
return cls.__instance
except AttributeError:
cls.__instance = cls()
return cls.__instance
def sync(self):
pass
class FakeLdap(object):
"""Fake LDAP connection."""
__prefix = 'ldap:'
def __init__(self, url):
LOG.debug('FakeLdap initialize url=%s', url)
if url == 'fake://memory':
self.db = FakeShelve.get_instance()
else:
self.db = shelve.open(url[7:])
def simple_bind_s(self, dn, password):
"""This method is ignored, but provided for compatibility."""
if server_fail:
raise ldap.SERVER_DOWN
LOG.debug('FakeLdap bind dn=%s', dn)
if dn == 'cn=Admin' and password == 'password':
return
try:
attrs = self.db['%s%s' % (self.__prefix, dn)]
except KeyError:
LOG.error('FakeLdap bind fail: dn=%s not found', dn)
raise ldap.NO_SUCH_OBJECT
db_password = None
try:
db_password = attrs['userPassword'][0]
except (KeyError, IndexError):
LOG.error('FakeLdap bind fail: password for dn=%s not found', dn)
raise ldap.INAPPROPRIATE_AUTH
if not utils.ldap_check_password(password, db_password):
LOG.error('FakeLdap bind fail: password for dn=%s does'
' not match' % dn)
raise ldap.INVALID_CREDENTIALS
def unbind_s(self):
"""This method is ignored, but provided for compatibility."""
if server_fail:
raise ldap.SERVER_DOWN
def add_s(self, dn, attrs):
"""Add an object with the specified attributes at dn."""
if server_fail:
raise ldap.SERVER_DOWN
key = '%s%s' % (self.__prefix, dn)
LOG.debug('FakeLdap add item: dn=%s, attrs=%s', dn, attrs)
if key in self.db:
LOG.error('FakeLdap add item failed: dn=%s is'
' already in store.', dn)
raise ldap.ALREADY_EXISTS(dn)
self.db[key] = dict([(k, v if isinstance(v, list) else [v])
for k, v in attrs])
self.db.sync()
def delete_s(self, dn):
"""Remove the ldap object at specified dn."""
if server_fail:
raise ldap.SERVER_DOWN
key = '%s%s' % (self.__prefix, dn)
LOG.debug('FakeLdap delete item: dn=%s', dn)
try:
del self.db[key]
except KeyError:
LOG.error('FakeLdap delete item failed: dn=%s not found.', dn)
raise ldap.NO_SUCH_OBJECT
self.db.sync()
def modify_s(self, dn, attrs):
"""Modify the object at dn using the attribute list.
:param dn: an LDAP DN
:param attrs: a list of tuples in the following form:
([MOD_ADD | MOD_DELETE | MOD_REPACE], attribute, value)
"""
if server_fail:
raise ldap.SERVER_DOWN
key = '%s%s' % (self.__prefix, dn)
LOG.debug('FakeLdap modify item: dn=%s attrs=%s', dn, attrs)
try:
entry = self.db[key]
except KeyError:
LOG.error('FakeLdap modify item failed: dn=%s not found.', dn)
raise ldap.NO_SUCH_OBJECT
for cmd, k, v in attrs:
values = entry.setdefault(k, [])
if cmd == ldap.MOD_ADD:
if isinstance(v, list):
values += v
else:
values.append(v)
elif cmd == ldap.MOD_REPLACE:
values[:] = v if isinstance(v, list) else [v]
elif cmd == ldap.MOD_DELETE:
if v is None:
if len(values) == 0:
LOG.error('FakeLdap modify item failed: '
'item has no attribute "%s" to delete', k)
raise ldap.NO_SUCH_ATTRIBUTE
values[:] = []
else:
if not isinstance(v, list):
v = [v]
for val in v:
try:
values.remove(val)
except ValueError:
LOG.error('FakeLdap modify item failed:'
' item has no attribute "%s" with'
' value "%s" to delete', k, val)
raise ldap.NO_SUCH_ATTRIBUTE
else:
LOG.error('FakeLdap modify item failed: unknown'
' command %s', cmd)
raise NotImplementedError('modify_s action %s not implemented'
% cmd)
self.db[key] = entry
self.db.sync()
def search_s(self, dn, scope, query=None, fields=None):
"""Search for all matching objects under dn using the query.
Args:
dn -- dn to search under
scope -- only SCOPE_BASE and SCOPE_SUBTREE are supported
query -- query to filter objects by
fields -- fields to return. Returns all fields if not specified
"""
if server_fail:
raise ldap.SERVER_DOWN
LOG.debug('FakeLdap search at dn=%s scope=%s query=%s',
dn, SCOPE_NAMES.get(scope, scope), query)
if scope == ldap.SCOPE_BASE:
try:
item_dict = self.db['%s%s' % (self.__prefix, dn)]
except KeyError:
LOG.debug('FakeLdap search fail: dn not found for SCOPE_BASE')
raise ldap.NO_SUCH_OBJECT
results = [(dn, item_dict)]
elif scope == ldap.SCOPE_SUBTREE:
results = [(k[len(self.__prefix):], v)
for k, v in self.db.iteritems()
if re.match('%s.*,%s' % (self.__prefix, dn), k)]
elif scope == ldap.SCOPE_ONELEVEL:
results = [(k[len(self.__prefix):], v)
for k, v in self.db.iteritems()
if re.match('%s\w+=[^,]+,%s' % (self.__prefix, dn), k)]
else:
LOG.error('FakeLdap search fail: unknown scope %s', scope)
raise NotImplementedError('Search scope %s not implemented.'
% scope)
objects = []
for dn, attrs in results:
# filter the objects by query
if not query or _match_query(query, attrs):
# filter the attributes by fields
attrs = dict([(k, v) for k, v in attrs.iteritems()
if not fields or k in fields])
objects.append((dn, attrs))
LOG.debug('FakeLdap search result: %s', objects)
return objects

15
keystone/common/utils.py
View File

@ -156,6 +156,21 @@ def hash_password(password):
return h
def ldap_hash_password(password):
"""Hash a password. Hard."""
password_utf8 = password.encode('utf-8')
h = passlib.hash.ldap_salted_sha1.encrypt(password_utf8)
return h
def ldap_check_password(password, hashed):
if password is None:
return False
password_utf8 = password.encode('utf-8')
h = passlib.hash.ldap_salted_sha1.encrypt(password_utf8)
return passlib.hash.ldap_salted_sha1.verify(password_utf8, hashed)
def check_password(password, hashed):
"""Check that a plaintext password matches hashed.

23
keystone/config.py
View File

@ -157,3 +157,26 @@ register_str('driver', group='identity')
register_str('driver', group='policy')
register_str('driver', group='token')
register_str('driver', group='ec2')
#ldap
register_str('url', group='ldap')
register_str('user', group='ldap')
register_str('password', group='ldap')
register_str('suffix', group='ldap')
register_bool('use_dumb_member', group='ldap')
register_str('user_tree_dn', group='ldap')
register_str('user_objectclass', group='ldap')
register_str('user_id_attribute', group='ldap')
register_str('tenant_tree_dn', group='ldap')
register_str('tenant_objectclass', group='ldap')
register_str('tenant_id_attribute', group='ldap')
register_str('tenant_member_attribute', group='ldap')
register_str('role_tree_dn', group='ldap')
register_str('role_objectclass', group='ldap')
register_str('role_id_attribute', group='ldap')
register_str('role_member_attribute', group='ldap')

1
keystone/identity/backends/ldap/__init__.py Normal file
View File

@ -0,0 +1 @@
from keystone.identity.backends.ldap.core import *

791
keystone/identity/backends/ldap/core.py Normal file
View File

@ -0,0 +1,791 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
import uuid
import ldap
from ldap import filter as ldap_filter
from keystone import config
from keystone import exception
from keystone import identity
from keystone.common import ldap as common_ldap
from keystone.common import utils
from keystone.common.ldap import fakeldap
from keystone.identity import models
CONF = config.CONF
def _filter_user(user_ref):
if user_ref:
user_ref.pop('password', None)
return user_ref
def _ensure_hashed_password(user_ref):
pw = user_ref.get('password', None)
if pw is not None:
pw = utils.ldap_hash_password(pw)
user_ref['password'] = pw
return user_ref
class Identity(identity.Driver):
def __init__(self):
super(Identity, self).__init__()
self.LDAP_URL = CONF.ldap.url
self.LDAP_USER = CONF.ldap.user
self.LDAP_PASSWORD = CONF.ldap.password
self.suffix = CONF.ldap.suffix
self.user = UserApi(CONF)
self.tenant = TenantApi(CONF)
self.role = RoleApi(CONF)
def get_connection(self, user=None, password=None):
if self.LDAP_URL.startswith('fake://'):
conn = fakeldap.FakeLdap(self.LDAP_URL)
else:
conn = common_ldap.LDAPWrapper(self.LDAP_URL)
if user is None:
user = self.LDAP_USER
if password is None:
password = self.LDAP_PASSWORD
conn.simple_bind_s(user, password)
return conn
# Identity interface
def authenticate(self, user_id=None, tenant_id=None, password=None):
"""Authenticate based on a user, tenant and password.
Expects the user object to have a password field and the tenant to be
in the list of tenants on the user.
"""
user_ref = self._get_user(user_id)
if user_ref is None:
raise AssertionError('Invalid user / password')
try:
conn = self.user.get_connection(self.user._id_to_dn(user_id),
password)
if not conn:
raise AssertionError('Invalid user / password')
except Exception:
raise AssertionError('Invalid user / password')
if tenant_id:
found = False
for tenant in user_ref['tenants']:
if tenant == tenant_id:
found = True
break
if not found:
raise AssertionError('Invalid tenant')
tenant_ref = self.tenant.get(tenant_id)
metadata_ref = {}
# TODO(termie): this should probably be made into a get roles call
#if tenant_ref:
# metadata_ref = self.get_metadata(user_id, tenant_id)
#else:
# metadata_ref = {}
return (_filter_user(user_ref), tenant_ref, metadata_ref)
def get_tenant(self, tenant_id):
return self.tenant.get(tenant_id)
def get_tenant_by_name(self, tenant_name):
return self.tenant.get_by_name(tenant_name)
def _get_user(self, user_id):
user_ref = self.user.get(user_id)
if not user_ref:
return None
tenants = self.tenant.get_user_tenants(user_id)
user_ref['tenants'] = []
for tenant in tenants:
user_ref['tenants'].append(tenant['id'])
return user_ref
def get_user(self, user_id):
user_ref = self._get_user(user_id)
if (not user_ref):
return None
return _filter_user(user_ref)
def get_metadata(self, user_id, tenant_id):
if not self.get_tenant(tenant_id):
return None
if not self.get_user(user_id):
return None
metadata_ref = self.get_roles_for_user_and_tenant(user_id, tenant_id)
return metadata_ref
def get_role(self, role_id):
return self.role.get(role_id)
# These should probably be part of the high-level API
def add_user_to_tenant(self, tenant_id, user_id):
return self.tenant.add_user(tenant_id, user_id)
def get_tenants_for_user(self, user_id):
tenant_list = []
for tenant in self.tenant.get_user_tenants(user_id):
tenant_list.append(tenant['id'])
return tenant_list
def get_roles_for_user_and_tenant(self, user_id, tenant_id):
assignments = self.role.get_role_assignments(tenant_id)
roles = []
for assignment in assignments:
if assignment.user_id == user_id:
roles.append(assignment.role_id)
return roles
def add_role_to_user_and_tenant(self, user_id, tenant_id, role_id):
self.role.add_user(role_id, user_id, tenant_id)
# CRUD
def create_user(self, user_id, user):
return self.user.create(user)
def update_user(self, user_id, user):
return self.user.update(user_id, user)
def create_tenant(self, tenant_id, tenant):
data = tenant.copy()
if 'id' not in data or data['id'] is None:
data['id'] = str(uuid.uuid4().hex)
return self.tenant.create(tenant)
def update_tenant(self, tenant_id, tenant):
return self.tenant.update(tenant_id, tenant)
def create_metadata(self, user_id, tenant_id, metadata):
return {}
def create_role(self, role_id, role):
return self.role.create(role)
def delete_role(self, role_id):
return self.role.delete(role_id)
# TODO(termie): remove this and move cross-api calls into driver
class ApiShim(object):
"""Quick singleton-y shim to get around recursive dependencies.
NOTE(termie): this should be removed and the cross-api code
should be moved into the driver itself.
"""
_role = None
_tenant = None
_user = None
def __init__(self, conf):
self.conf = conf
@property
def role(self):
if not self._role:
self._role = RoleApi(self.conf)
return self._role
@property
def tenant(self):
if not self._tenant:
self._tenant = TenantApi(self.conf)
return self._tenant
@property
def user(self):
if not self._user:
self._user = UserApi(self.conf)
return self._user
# TODO(termie): remove this and move cross-api calls into driver
class ApiShimMixin(object):
"""Mixin to share some ApiShim code. Remove me."""
@property
def role_api(self):
return self.api.role
@property
def tenant_api(self):
return self.api.tenant
@property
def user_api(self):
return self.api.user
# TODO(termie): turn this into a data object and move logic to driver
class UserApi(common_ldap.BaseLdap, ApiShimMixin):
DEFAULT_OU = 'ou=Users'
DEFAULT_STRUCTURAL_CLASSES = ['person']
DEFAULT_ID_ATTRIBUTE = 'cn'
DEFAULT_OBJECTCLASS = 'inetOrgPerson'
options_name = 'user'
attribute_mapping = {'password': 'userPassword',
#'email': 'mail',
'name': 'sn'}
# NOTE(ayoung): The RFC based schemas don't have a way to indicate
# 'enabled' the closest is the nsAccount lock, which is on defined to
# be part of any objectclass.
# in the future, we need to provide a way for the end user to
# indicate the field to use and what it indicates
attribute_ignore = ['tenant_id', 'enabled', 'tenants']
model = models.User
def __init__(self, conf):
super(UserApi, self).__init__(conf)
self.api = ApiShim(conf)
def get_by_name(self, name, filter=None):
users = self.get_all('(%s=%s)' %
(self.attribute_mapping['name'],
ldap_filter.escape_filter_chars(name)))
try:
return users[0]
except IndexError:
return None
def create(self, values):
self.affirm_unique(values)
_ensure_hashed_password(values)
values = super(UserApi, self).create(values)
tenant_id = values.get('tenant_id')
if tenant_id is not None:
self.tenant_api.add_user(values['tenant_id'], values['id'])
return values
def update(self, id, values):
if values['id'] != id:
return None
old_obj = self.get(id)
if old_obj.get('name') != values['name']:
raise exception.Error('Changing Name not permitted')
try:
new_tenant = values['tenant_id']
except KeyError:
pass
else:
if old_obj.get('tenant_id') != new_tenant:
if old_obj['tenant_id']:
self.tenant_api.remove_user(old_obj['tenant_id'], id)
if new_tenant:
self.tenant_api.add_user(new_tenant, id)
_ensure_hashed_password(values)
super(UserApi, self).update(id, values, old_obj)
def delete(self, id):
user = self.get(id)
if user.tenant_id:
self.tenant_api.remove_user(user.tenant_id, id)
super(UserApi, self).delete(id)
for ref in self.role_api.list_global_roles_for_user(id):
self.role_api.rolegrant_delete(ref.id)
for ref in self.role_api.list_tenant_roles_for_user(id):
self.role_api.rolegrant_delete(ref.id)
def get_by_email(self, email):
users = self.get_all('(mail=%s)' % \
(ldap_filter.escape_filter_chars(email),))
try:
return users[0]
except IndexError:
return None
def user_roles_by_tenant(self, user_id, tenant_id):
return self.role_api.list_tenant_roles_for_user(user_id, tenant_id)
def get_by_tenant(self, user_id, tenant_id):
user_dn = self._id_to_dn(user_id)
user = self.get(user_id)
tenant = self.tenant_api._ldap_get(tenant_id,
'(member=%s)' % (user_dn,))
if tenant is not None:
return user
else:
if self.role_api.list_tenant_roles_for_user(user_id, tenant_id):
return user
return None
def user_role_add(self, values):
return self.role_api.add_user(values.role_id, values.user_id,
values.tenant_id)
def users_get_page(self, marker, limit):
return self.get_page(marker, limit)
def users_get_page_markers(self, marker, limit):
return self.get_page_markers(marker, limit)
def users_get_by_tenant_get_page(self, tenant_id, role_id, marker, limit):
return self._get_page(marker,
limit,
self.tenant_api.get_users(tenant_id, role_id))
def users_get_by_tenant_get_page_markers(self, tenant_id,
role_id, marker, limit):
return self._get_page_markers(
marker, limit, self.tenant_api.get_users(tenant_id, role_id))
def check_password(self, user_id, password):
user = self.get(user_id)
return utils.check_password(password, user.password)
# TODO(termie): turn this into a data object and move logic to driver
class TenantApi(common_ldap.BaseLdap, ApiShimMixin):
DEFAULT_OU = 'ou=Groups'
DEFAULT_STRUCTURAL_CLASSES = []
DEFAULT_OBJECTCLASS = 'groupOfNames'
DEFAULT_ID_ATTRIBUTE = 'cn'
DEFAULT_MEMBER_ATTRIBUTE = 'member'
options_name = 'tenant'
attribute_mapping = {
#'description': 'desc', 'enabled': 'keystoneEnabled',
'name': 'ou'}
model = models.Tenant
def __init__(self, conf):
super(TenantApi, self).__init__(conf)
self.api = ApiShim(conf)
self.member_attribute = (getattr(conf.ldap, 'tenant_member_attribute')
or self.DEFAULT_MEMBER_ATTRIBUTE)
def get_by_name(self, name, filter=None): # pylint: disable=W0221,W0613
search_filter = ('(%s=%s)'
% (self.attribute_mapping['name'],
ldap_filter.escape_filter_chars(name)))
tenants = self.get_all(search_filter)
try:
return tenants[0]
except IndexError:
return None
def create(self, values):
self.affirm_unique(values)
data = values.copy()
if 'id' not in data or data['id'] is None:
data['id'] = uuid.uuid4().hex
return super(TenantApi, self).create(data)
def get_user_tenants(self, user_id):
"""Returns list of tenants a user has access to
Always includes default tenants.
"""
user_dn = self.user_api._id_to_dn(user_id)
query = '(%s=%s)' % (self.member_attribute, user_dn)
memberships = self.get_all(query)
return memberships
def list_for_user_get_page(self, user, marker, limit):
return self._get_page(marker, limit, self.get_user_tenants(user['id']))
def list_for_user_get_page_markers(self, user, marker, limit):
return self._get_page_markers(
marker, limit, self.get_user_tenants(user['id']))
def is_empty(self, id):
tenant = self._ldap_get(id)
members = tenant[1].get(self.member_attribute, [])
if self.use_dumb_member:
empty = members == [self.DUMB_MEMBER_DN]
else:
empty = len(members) == 0
return empty and len(self.role_api.get_role_assignments(id)) == 0
def get_role_assignments(self, tenant_id):
return self.role_api.get_role_assignments(tenant_id)
def add_user(self, tenant_id, user_id):
conn = self.get_connection()
conn.modify_s(self._id_to_dn(tenant_id),
[(ldap.MOD_ADD,
self.member_attribute,
self.user_api._id_to_dn(user_id))])
def remove_user(self, tenant_id, user_id):
conn = self.get_connection()
conn.modify_s(self._id_to_dn(tenant_id),
[(ldap.MOD_DELETE,
self.member_attribute,
self.user_api._id_to_dn(user_id))])
def get_users(self, tenant_id, role_id=None):
tenant = self._ldap_get(tenant_id)
res = []
if not role_id:
# Get users who have default tenant mapping
for user_dn in tenant[1].get(self.member_attribute, []):
if self.use_dumb_member and user_dn == self.DUMB_MEMBER_DN:
continue
res.append(self.user_api.get(self.user_api._dn_to_id(user_dn)))
# Get users who are explicitly mapped via a tenant
rolegrants = self.role_api.get_role_assignments(tenant_id)
for rolegrant in rolegrants:
if role_id is None or rolegrant.role_id == role_id:
res.append(self.user_api.get(rolegrant.user_id))
return res
def delete(self, id):
super(TenantApi, self).delete(id)
def update(self, id, values):
old_obj = self.get(id)
if old_obj['name'] != values['name']:
raise exception.Error('Changing Name not permitted')
super(TenantApi, self).update(id, values, old_obj)
class UserRoleAssociation(object):
"""Role Grant model."""
hints = {
'contract_attributes': ['id', 'role_id', 'user_id', 'tenant_id'],
'types': [('user_id', basestring), ('tenant_id', basestring)],
'maps': {'userId': 'user_id',
'roleId': 'role_id',
'tenantId': 'tenant_id'}
}
def __init__(self, user_id=None, role_id=None, tenant_id=None,
*args, **kw):
self.user_id = str(user_id)
self.role_id = role_id
self.tenant_id = str(tenant_id)
# TODO(termie): turn this into a data object and move logic to driver
class RoleApi(common_ldap.BaseLdap, ApiShimMixin):
DEFAULT_OU = 'ou=Roles'
DEFAULT_STRUCTURAL_CLASSES = []
options_name = 'role'
DEFAULT_OBJECTCLASS = 'organizationalRole'
DEFAULT_MEMBER_ATTRIBUTE = 'roleOccupant'
attribute_mapping = {'name': 'cn',
#'serviceId': 'service_id',
}
model = models.Tenant
def __init__(self, conf):
super(RoleApi, self).__init__(conf)
self.api = ApiShim(conf)
self.member_attribute = (getattr(conf.ldap, 'role_member_attribute')
or self.DEFAULT_MEMBER_ATTRIBUTE)
@staticmethod
def _create_ref(role_id, tenant_id, user_id):
role_id = '' if role_id is None else str(role_id)
tenant_id = '' if tenant_id is None else str(tenant_id)
user_id = '' if user_id is None else str(user_id)
return '%d-%d-%s%s%s' % (len(role_id),
len(tenant_id),
role_id,
tenant_id,
user_id)
@staticmethod
def _explode_ref(rolegrant):
a = rolegrant.split('-', 2)
len_role = int(a[0])
len_tenant = int(a[1])
role_id = a[2][:len_role]
role_id = None if len(role_id) == 0 else str(role_id)
tenant_id = a[2][len_role:len_tenant + len_role]
tenant_id = None if len(tenant_id) == 0 else str(tenant_id)
user_id = a[2][len_tenant + len_role:]
user_id = None if len(user_id) == 0 else str(user_id)
return role_id, tenant_id, user_id
def _subrole_id_to_dn(self, role_id, tenant_id):
if tenant_id is None:
return self._id_to_dn(role_id)
else:
return 'cn=%s,%s' % (ldap.dn.escape_dn_chars(role_id),
self.tenant_api._id_to_dn(tenant_id))
def get(self, id, filter=None):
model = super(RoleApi, self).get(id, filter)
return model
def create(self, values):
#values['id'] = values['name']
#delattr(values, 'name')
return super(RoleApi, self).create(values)
# pylint: disable=W0221
def get_by_name(self, name, filter=None):
return self.get(name, filter)
def add_user(self, role_id, user_id, tenant_id=None):
user = self.user_api.get(user_id)
if user is None:
raise exception.NotFound('User %s not found' % (user_id,))
role_dn = self._subrole_id_to_dn(role_id, tenant_id)
conn = self.get_connection()
user_dn = self.user_api._id_to_dn(user_id)
try:
conn.modify_s(role_dn, [(ldap.MOD_ADD,
self.member_attribute, user_dn)])
except ldap.TYPE_OR_VALUE_EXISTS:
raise exception.Error('User %s already has role %s in tenant %s'
% (user_id, role_id, tenant_id))
except ldap.NO_SUCH_OBJECT:
if tenant_id is None or self.get(role_id) is None:
raise Exception("Role %s not found" % (role_id,))
attrs = [('objectClass', [self.object_class]),
(self.member_attribute, [user_dn])]
if self.use_dumb_member:
attrs[1][1].append(self.DUMB_MEMBER_DN)
try:
conn.add_s(role_dn, attrs)
except Exception as inst:
raise inst
return UserRoleAssociation(
id=self._create_ref(role_id, tenant_id, user_id),
role_id=role_id,
user_id=user_id,
tenant_id=tenant_id)
def get_by_service(self, service_id):
roles = self.get_all('(service_id=%s)' %
ldap_filter.escape_filter_chars(service_id))
try:
res = []
for role in roles:
res.append(role)
return res
except IndexError:
return None
def get_role_assignments(self, tenant_id):
conn = self.get_connection()
query = '(objectClass=%s)' % self.object_class
tenant_dn = self.tenant_api._id_to_dn(tenant_id)
try:
roles = conn.search_s(tenant_dn, ldap.SCOPE_ONELEVEL, query)
except ldap.NO_SUCH_OBJECT:
return []
res = []
for role_dn, attrs in roles:
try:
user_dns = attrs[self.member_attribute]
except KeyError:
continue
for user_dn in user_dns:
if self.use_dumb_member and user_dn == self.DUMB_MEMBER_DN:
continue
user_id = self.user_api._dn_to_id(user_dn)
role_id = self._dn_to_id(role_dn)
res.append(UserRoleAssociation(
id=self._create_ref(role_id, tenant_id, user_id),
user_id=user_id,
role_id=role_id,
tenant_id=tenant_id))
return res
def list_global_roles_for_user(self, user_id):
user_dn = self.user_api._id_to_dn(user_id)
roles = self.get_all('(%s=%s)' % (self.member_attribute, user_dn))
return [UserRoleAssociation(
id=self._create_ref(role.id, None, user_id),
role_id=role.id,
user_id=user_id)
for role in roles]
def list_tenant_roles_for_user(self, user_id, tenant_id=None):
conn = self.get_connection()
user_dn = self.user_api._id_to_dn(user_id)
query = '(&(objectClass=%s)(%s=%s))' % (self.object_class,
self.member_attribute,
user_dn)
if tenant_id is not None:
tenant_dn = self.tenant_api._id_to_dn(tenant_id)
try:
roles = conn.search_s(tenant_dn, ldap.SCOPE_ONELEVEL, query)
except ldap.NO_SUCH_OBJECT:
return []
res = []
for role_dn, _ in roles:
role_id = self._dn_to_id(role_dn)
res.append(UserRoleAssociation(
id=self._create_ref(role_id, tenant_id, user_id),
user_id=user_id,
role_id=role_id,
tenant_id=tenant_id))
else:
try:
roles = conn.search_s(self.tenant_api.tree_dn,
ldap.SCOPE_SUBTREE,
query)
except ldap.NO_SUCH_OBJECT:
return []
res = []
for role_dn, _ in roles:
role_id = self._dn_to_id(role_dn)
tenant_id = ldap.dn.str2dn(role_dn)[1][0][1]
res.append(UserRoleAssociation(
id=self._create_ref(role_id, tenant_id, user_id),
user_id=user_id,
role_id=role_id,
tenant_id=tenant_id))
return res
def rolegrant_get(self, id):
role_id, tenant_id, user_id = self._explode_ref(id)
user_dn = self.user_api._id_to_dn(user_id)
role_dn = self._subrole_id_to_dn(role_id, tenant_id)
query = '(&(objectClass=%s)(%s=%s))' % (self.object_class,
self.member_attribute,
user_dn)
conn = self.get_connection()
try:
res = conn.search_s(role_dn, ldap.SCOPE_BASE, query)
except ldap.NO_SUCH_OBJECT:
return None
if len(res) == 0:
return None
return UserRoleAssociation(id=id,
role_id=role_id,
tenant_id=tenant_id,
user_id=user_id)
def rolegrant_delete(self, id):
role_id, tenant_id, user_id = self._explode_ref(id)
user_dn = self.user_api._id_to_dn(user_id)
role_dn = self._subrole_id_to_dn(role_id, tenant_id)
conn = self.get_connection()
try:
conn.modify_s(role_dn, [(ldap.MOD_DELETE, '', [user_dn])])
except ldap.NO_SUCH_ATTRIBUTE:
raise exception.Error("No such user in role")
def rolegrant_get_page(self, marker, limit, user_id, tenant_id):
all_roles = []
if tenant_id is None:
all_roles += self.list_global_roles_for_user(user_id)
else:
for tenant in self.tenant_api.get_all():
all_roles += self.list_tenant_roles_for_user(user_id,
tenant['id'])
return self._get_page(marker, limit, all_roles)
def rolegrant_get_page_markers(self, user_id, tenant_id, marker, limit):
all_roles = []
if tenant_id is None:
all_roles = self.list_global_roles_for_user(user_id)
else:
for tenant in self.tenant_api.get_all():
all_roles += self.list_tenant_roles_for_user(user_id,
tenant['id'])
return self._get_page_markers(marker, limit, all_roles)
def get_by_service_get_page(self, service_id, marker, limit):
all_roles = self.get_by_service(service_id)
return self._get_page(marker, limit, all_roles)
def get_by_service_get_page_markers(self, service_id, marker, limit):
all_roles = self.get_by_service(service_id)
return self._get_page_markers(marker, limit, all_roles)
def rolegrant_list_by_role(self, id):
role_dn = self._id_to_dn(id)
try:
roles = self.get_all('(%s=%s)' % (self.member_attribute, role_dn))
except ldap.NO_SUCH_OBJECT:
return []
res = []
for role_dn, attrs in roles:
try:
user_dns = attrs[self.member_attribute]
tenant_dns = attrs['tenant']
except KeyError:
continue
for user_dn in user_dns:
if self.use_dumb_member and user_dn == self.DUMB_MEMBER_DN:
continue
user_id = self.user_api._dn_to_id(user_dn)
tenant_id = None
if tenant_dns is not None:
for tenant_dn in tenant_dns:
tenant_id = self.tenant_api._dn_to_id(tenant_dn)
role_id = self._dn_to_id(role_dn)
res.append(UserRoleAssociation(
id=self._create_ref(role_id, tenant_id, user_id),
user_id=user_id,
role_id=role_id,
tenant_id=tenant_id))
return res
def rolegrant_get_by_ids(self, user_id, role_id, tenant_id):
conn = self.get_connection()
user_dn = self.user_api._id_to_dn(user_id)
query = '(&(objectClass=%s)(%s=%s))' % (self.object_class,
self.member_attribute,
user_dn)
if tenant_id is not None:
tenant_dn = self.tenant_api._id_to_dn(tenant_id)
try:
roles = conn.search_s(tenant_dn, ldap.SCOPE_ONELEVEL, query)
except ldap.NO_SUCH_OBJECT:
return None
if len(roles) == 0:
return None
for role_dn, _ in roles:
ldap_role_id = self._dn_to_id(role_dn)
if role_id == ldap_role_id:
res = UserRoleAssociation(
id=self._create_ref(role_id, tenant_id, user_id),
user_id=user_id,
role_id=role_id,
tenant_id=tenant_id)
return res
else:
try:
roles = self.get_all('(%s=%s)' % (self.member_attribute,
user_dn))
except ldap.NO_SUCH_OBJECT:
return None
if len(roles) == 0:
return None
for role in roles:
if role.id == role_id:
return UserRoleAssociation(
id=self._create_ref(role.id, None, user_id),
role_id=role.id,
user_id=user_id)
return None

6
keystone/identity/core.py
View File

@ -143,7 +143,7 @@ class Driver(object):
"""
raise NotImplementedError()
def add_role_for_user_and_tenant(self, user_id, tenant_id, role_id):
def add_role_to_user_and_tenant(self, user_id, tenant_id, role_id):
"""Add a role to a user within given tenant."""
raise NotImplementedError()
@ -172,6 +172,10 @@ class Driver(object):
raise NotImplementedError()
# metadata crud
def get_metadata(self, user_id, tenant_id):
raise NotImplementedError()
def create_metadata(self, user_id, tenant_id, metadata):
raise NotImplementedError()

77
keystone/identity/models.py Normal file
View File

@ -0,0 +1,77 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright (C) 2011 OpenStack LLC.
#
# 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.
"""Model descriptions.
Unless marked otherwise, all fields are strings.
"""
class Model(dict):
"""Base model class."""
@property
def known_keys(cls):
return cls.required_keys + cls.optional_keys
class User(Model):
"""User object.
Required keys:
id
name
Optional keys:
password
description
email
enabled (bool, default True)
"""
required_keys = ('id', 'name')
optional_keys = ('password', 'description', 'email', 'enabled')
class Tenant(Model):
"""Tenant object.
Required keys:
id
name
Optional Keys:
description
enabled (bool, default True)
"""
required_keys = ('id', 'name')
optional_keys = ('description', 'enabled')
class Role(Model):
"""Role object.
Required keys:
id
name
"""
required_keys = ('id', 'name')
optional_keys = tuple()

5
keystone/test.py
View File

@ -233,6 +233,11 @@ class TestCase(unittest.TestCase):
return super(TestCase, self).assertIsNotNone(actual)
self.assert_(actual is not None)
def assertIsNone(self, actual):
if hasattr(super(TestCase, self), 'assertIsNone'):
return super(TestCase, self).assertIsNone(actual)
self.assert_(actual is None)
def assertNotIn(self, needle, haystack):
if hasattr(super(TestCase, self), 'assertNotIn'):
return super(TestCase, self).assertNotIn(needle, haystack)

61
tests/_ldap_livetest.py Normal file
View File

@ -0,0 +1,61 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
import subprocess
from keystone import config
from keystone import test
from keystone.identity.backends import ldap as identity_ldap
import default_fixtures
import test_backend
CONF = config.CONF
def delete_object(name):
devnull = open('/dev/null', 'w')
dn = '%s,%s' % (name, CONF.ldap.suffix)
subprocess.call(['ldapdelete',
'-x',
'-D', CONF.ldap.user,
'-H', CONF.ldap.url,
'-w', CONF.ldap.password,
dn],
stderr=devnull)
def clear_live_database():
roles = ['keystone_admin']
groups = ['baz', 'bar', 'tenent4add','fake1','fake2']
users = ['foo', 'two','fake1','fake2']
roles = ['keystone_admin', 'useless']
for group in groups:
for role in roles:
delete_object ('cn=%s,cn=%s,ou=Groups' % (role, group))
delete_object('cn=%s,ou=Groups' % group)
for user in users:
delete_object ('cn=%s,ou=Users' % user)
for role in roles:
delete_object ('cn=%s,ou=Roles' % role)
class LDAPIdentity(test.TestCase, test_backend.IdentityTests):
def setUp(self):
super(LDAPIdentity, self).setUp()
CONF(config_files=[test.etcdir('keystone.conf'),
test.testsdir('test_overrides.conf'),
test.testsdir('backend_liveldap.conf')])
clear_live_database()
self.identity_api = identity_ldap.Identity()
self.load_fixtures(default_fixtures)
self.user_foo = {'id': 'foo',
'name': 'FOO',
'password': 'foo2',
'tenants': ['bar']}
def tearDown(self):
test.TestCase.tearDown(self)

9
tests/backend_ldap.conf Normal file
View File

@ -0,0 +1,9 @@
[ldap]
url = fake://memory
user = cn=Admin
password = password
backend_entities = ['Tenant', 'User', 'UserRoleAssociation', 'Role']
tree_dn = cn=example,cn=com
[identity]
driver = keystone.identity.backends.ldap.Identity

9
tests/backend_liveldap.conf Normal file
View File

@ -0,0 +1,9 @@
[ldap]
url = ldap://localhost
suffix = dc=younglogic,dc=com
user_tree_dn = ou=Users,dc=younglogic,dc=com
role_tree_dn = ou=Roles,dc=younglogic,dc=com
tenant_tree_dn = ou=Groups,dc=younglogic,dc=com
user = dc=Manager,dc=younglogic,dc=com
password = freeipa4all
backend_entities = ['Tenant', 'User', 'UserRoleAssociation', 'Role']

1
tests/default_fixtures.py
View File

@ -1,6 +1,7 @@
TENANTS = [
{'id': 'bar', 'name': 'BAR'},
{'id': 'baz', 'name': 'BAZ'},
{'id': 'tenent4add', 'name': 'tenant4add'},
]
# NOTE(ja): a role of keystone_admin and attribute "is_admin" is done in setUp

39
tests/test_backend.py
View File

@ -78,7 +78,8 @@ class IdentityTests(object):
self.assert_(tenant_ref is None)
def test_get_tenant(self):
tenant_ref = self.identity_api.get_tenant(tenant_id=self.tenant_bar['id'])
tenant_ref = self.identity_api.get_tenant(
tenant_id=self.tenant_bar['id'])
self.assertDictEquals(tenant_ref, self.tenant_bar)
def test_get_tenant_by_name_bad_tenant(self):
@ -221,6 +222,42 @@ class IdentityTests(object):
self.assert_(tenant_ref is None)
def test_get_role_by_user_and_tenant(self):
roles_ref = self.identity_api.get_roles_for_user_and_tenant(
self.user_foo['id'],self.tenant_bar['id'])
self.assertNotIn('keystone_admin', roles_ref)
self.identity_api.add_role_to_user_and_tenant(
self.user_foo['id'],self.tenant_bar['id'], 'keystone_admin')
roles_ref = self.identity_api.get_roles_for_user_and_tenant(
self.user_foo['id'],self.tenant_bar['id'])
self.assertIn('keystone_admin', roles_ref)
self.assertNotIn('useless',roles_ref)
self.identity_api.add_role_to_user_and_tenant(
self.user_foo['id'],self.tenant_bar['id'], 'useless')
roles_ref = self.identity_api.get_roles_for_user_and_tenant(
self.user_foo['id'],self.tenant_bar['id'])
self.assertIn('keystone_admin', roles_ref)
self.assertIn('useless',roles_ref)
def test_delete_role(self):
role_id = 'test_role_delete'
new_role = {'id': role_id, 'name': 'Role to Delete'}
self.identity_api.create_role(role_id , new_role)
role_ref = self.identity_api.get_role(role_id)
self.assertDictEquals(role_ref, new_role)
self.identity_api.delete_role(role_id)
role_ref = self.identity_api.get_role(role_id)
print role_ref
self.assertIsNone(role_ref)
def test_add_user_to_tenant(self):
tenant_id = 'tenent4add'
self.identity_api.add_user_to_tenant(tenant_id, 'foo')
tenants = self.identity_api.get_tenants_for_user('foo')
self.assertIn(tenant_id, tenants)
class TokenTests(object):
def test_token_crud(self):
token_id = uuid.uuid4().hex

35
tests/test_backend_ldap.py Normal file
View File

@ -0,0 +1,35 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
from keystone import config
from keystone import test
from keystone.common.ldap import fakeldap
from keystone.identity.backends import ldap as identity_ldap
import default_fixtures
import test_backend
CONF = config.CONF
def clear_database():
db = fakeldap.FakeShelve().get_instance()
db.clear()
class LDAPIdentity(test.TestCase, test_backend.IdentityTests):
def setUp(self):
super(LDAPIdentity, self).setUp()
CONF(config_files=[test.etcdir('keystone.conf'),
test.testsdir('test_overrides.conf'),
test.testsdir('backend_ldap.conf')])
clear_database()
self.identity_api = identity_ldap.Identity()
self.load_fixtures(default_fixtures)
self.user_foo = {'id': 'foo',
'name': 'FOO',
'password': 'foo2',
'tenants': ['bar']}
def tearDown(self):
test.TestCase.tearDown(self)

2
tests/test_keystoneclient.py
View File

@ -221,7 +221,7 @@ class KeystoneClientTests(object):
# Admin endpoint should return *all* tenants
client = self.get_client(admin=True)
tenants = client.tenants.list()
self.assertEquals(len(tenants), 2)
self.assertEquals(len(tenants), len(default_fixtures.TENANTS))
def test_invalid_password(self):
from keystoneclient import exceptions as client_exceptions

1
tools/pip-requires
View File

@ -31,3 +31,4 @@ mox # mock object framework
-e git+https://review.openstack.org/p/openstack/python-keystoneclient.git#egg=python-keystoneclient
-e git+https://review.openstack.org/p/openstack-dev/openstack-nose.git#egg=openstack.nose_plugin
python-ldap==2.3.13# authenticate against an existing LDAP server

4
tools/pip-requires-test
View File

@ -2,6 +2,7 @@
pam==0.1.4
WebOb==1.0.8
eventlet
greenlet
PasteDeploy
paste
routes
@ -22,3 +23,6 @@ pep8
# for python-novaclient
prettytable
# Optional backend: LDAP
python-ldap==2.3.13 # authenticate against an existing LDAP server