OpenStack Identity (Keystone)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

645 lines
23 KiB

# 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
# 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 keystone.
import random
import re
import shelve
import ldap
from oslo_log import log
import keystone.conf
from keystone import exception
from keystone.identity.backends.ldap import common
LOG = log.getLogger(__name__)
CONF = keystone.conf.CONF
def _internal_attr(attr_name, value_or_values):
def normalize_value(value):
return common.utf8_decode(value)
def normalize_dn(dn):
# Capitalize the attribute names as an LDAP server might.
# NOTE(blk-u): Special case for this tested value, used with
# test_user_id_comma. The call to str2dn here isn't always correct
# here, because `dn` is escaped for an LDAP filter. str2dn() normally
# works only because there's no special characters in `dn`.
if dn == 'cn=Doe\\5c, John,ou=Users,cn=example,cn=com':
return 'CN=Doe\\, John,OU=Users,CN=example,CN=com'
# NOTE(blk-u): Another special case for this tested value. When a
# roleOccupant has an escaped comma, it gets converted to \2C.
if dn == 'cn=Doe\\, John,ou=Users,cn=example,cn=com':
return 'CN=Doe\\2C John,OU=Users,CN=example,CN=com'
dn = ldap.dn.str2dn(dn)
except ldap.DECODING_ERROR:
# NOTE(amakarov): In case of IDs instead of DNs in group members
# they must be handled as regular values.
return normalize_value(dn)
norm = []
for part in dn:
name, val, i = part[0]
name = name.upper()
norm.append([(name, val, i)])
return ldap.dn.dn2str(norm)
if attr_name in ('member', 'roleOccupant'):
attr_fn = normalize_dn
attr_fn = normalize_value
if isinstance(value_or_values, list):
return [attr_fn(x) for x in value_or_values]
return [attr_fn(value_or_values)]
def _match_query(query, attrs, attrs_checked):
"""Match an ldap query to an attribute dictionary.
The characters &, |, and ! are supported in the query. No syntax checking
is performed, so malformed queries will not work correctly.
# cut off the parentheses
inner = query[1:-1]
if inner.startswith(('&', '|')):
if inner[0] == '&':
matchfn = all
matchfn = any
# cut off the & or |
groups = _paren_groups(inner[1:])
return matchfn(_match_query(group, attrs, attrs_checked)
for group in groups)
if inner.startswith('!'):
# cut off the ! and the nested parentheses
return not _match_query(query[2:-1], attrs, attrs_checked)
(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 range(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."""
def match_with_wildcards(norm_val, val_list):
# Case insensitive checking with wildcards
if norm_val.startswith('*'):
if norm_val.endswith('*'):
# Is the string anywhere in the target?
for x in val_list:
if norm_val[1:-1] in x:
return True
# Is the string at the end of the target?
for x in val_list:
if (norm_val[1:] ==
x[len(x) - len(norm_val) + 1:]):
return True
elif norm_val.endswith('*'):
# Is the string at the start of the target?
for x in val_list:
if norm_val[:-1] == x[:len(norm_val) - 1]:
return True
# Is the string an exact match?
for x in val_list:
if check_value == x:
return True
return False
if key not in attrs:
return False
# This is a pure wild card search, so the answer must be yes!
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':
check_value = _internal_attr(key, value)[0].lower()
norm_values = list(
_internal_attr(key, x)[0].lower() for x in attrs[key])
return match_with_wildcards(check_value, norm_values)
# 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):
"""Return 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': ['keystoneProject',
if value in subs:
return [value] + subs[value]
return [value]
server_fail = False
class FakeShelve(dict):
def sync(self):
FakeShelves = {}
PendingRequests = {}
class FakeLdap(common.LDAPHandler):
"""Emulate the python-ldap API.
The python-ldap API requires all strings to be UTF-8 encoded with the
exception of [1]. This is assured by the caller of this interface
(i.e. KeystoneLDAPHandler).
However, internally this emulation MUST process and store strings
in a canonical form which permits operations on
characters. Encoded strings do not provide the ability to operate
on characters. Therefore this emulation accepts UTF-8 encoded
strings, decodes them to unicode for operations internal to this
emulation, and encodes them back to UTF-8 when returning values
from the emulation.
[1] Some fields (DNs, RDNs, attribute names, queries) are represented
as text in python-ldap for Python 3, and for Python 2 when
bytes_mode=False. For more details see:
__prefix = 'ldap:'
def __init__(self, conn=None):
super(FakeLdap, self).__init__(conn=conn)
self._ldap_options = {ldap.OPT_DEREF: ldap.DEREF_NEVER}
def connect(self, url, page_size=0, alias_dereferencing=None,
use_tls=False, tls_cacertfile=None, tls_cacertdir=None,
tls_req_cert='demand', chase_referrals=None, debug_level=None,
use_pool=None, pool_size=None, pool_retry_max=None,
pool_retry_delay=None, pool_conn_timeout=None,
if url.startswith('fake://memory'):
if url not in FakeShelves:
FakeShelves[url] = FakeShelve()
self.db = FakeShelves[url]
self.db =[7:])
using_ldaps = url.lower().startswith("ldaps")
if use_tls and using_ldaps:
raise AssertionError('Invalid TLS / LDAPS combination')
if use_tls:
if tls_cacertfile:
ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, tls_cacertfile)
elif tls_cacertdir:
ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, tls_cacertdir)
if tls_req_cert in list(common.LDAP_TLS_CERTS.values()):
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, tls_req_cert)
raise ValueError("invalid TLS_REQUIRE_CERT tls_req_cert=%s",
if alias_dereferencing is not None:
self.set_option(ldap.OPT_DEREF, alias_dereferencing)
self.page_size = page_size
self.use_pool = use_pool
self.pool_size = pool_size
self.pool_retry_max = pool_retry_max
self.pool_retry_delay = pool_retry_delay
self.pool_conn_timeout = pool_conn_timeout
self.pool_conn_lifetime = pool_conn_lifetime
self.conn_timeout = conn_timeout
def _dn_to_id_attr(self, dn):
return ldap.dn.str2dn(dn)[0][0][0]
def _dn_to_id_value(self, dn):
return ldap.dn.str2dn(dn)[0][0][1]
def key(self, dn):
return '%s%s' % (self.__prefix, dn)
def simple_bind_s(self, who='', cred='',
serverctrls=None, clientctrls=None):
"""Provide for compatibility but this method is ignored."""
if server_fail:
raise ldap.SERVER_DOWN
whos = ['cn=Admin', CONF.ldap.user]
if (who in whos and cred in ['password', CONF.ldap.password]):
attrs = self.db.get(self.key(who))
if not attrs:
LOG.debug('who=%s not found, binding anonymously', who)
db_password = ''
if attrs:
db_password = attrs['userPassword'][0]
except (KeyError, IndexError):
LOG.debug('bind fail: password for who=%s not found', who)
if cred != db_password:
LOG.debug('bind fail: password for who=%s does not match', who)
def unbind_s(self):
"""Provide for compatibility but this method is ignored."""
if server_fail:
raise ldap.SERVER_DOWN
def add_s(self, dn, modlist):
"""Add an object with the specified attributes at dn."""
if server_fail:
raise ldap.SERVER_DOWN
id_attr_in_modlist = False
id_attr = self._dn_to_id_attr(dn)
id_value = self._dn_to_id_value(dn)
# The LDAP API raises a TypeError if attr name is None.
for k, dummy_v in modlist:
if k is None:
raise TypeError('must be string, not None. modlist=%s' %
if k == id_attr:
for val in dummy_v:
if common.utf8_decode(val) == id_value:
id_attr_in_modlist = True
if not id_attr_in_modlist:
LOG.debug('id_attribute=%(attr)s missing, attributes=%(attrs)s',
{'attr': id_attr, 'attrs': modlist})
key = self.key(dn)
LOG.debug('add item: dn=%(dn)s, attrs=%(attrs)s', {
'dn': dn, 'attrs': modlist})
if key in self.db:
LOG.debug('add item failed: dn=%s is already in store.', dn)
raise ldap.ALREADY_EXISTS(dn)
self.db[key] = {k: _internal_attr(k, v) for k, v in modlist}
def delete_s(self, dn):
"""Remove the ldap object at specified dn."""
return self.delete_ext_s(dn, serverctrls=[])
def _getChildren(self, dn):
return [k for k, v in self.db.items()
if re.match('%s.*,%s' % (
re.escape(dn)), k)]
def delete_ext_s(self, dn, serverctrls, clientctrls=None):
"""Remove the ldap object at specified dn."""
if server_fail:
raise ldap.SERVER_DOWN
key = self.key(dn)
LOG.debug('FakeLdap delete item: dn=%s', dn)
del self.db[key]
except KeyError:
LOG.debug('delete item failed: dn=%s not found.', dn)
raise ldap.NO_SUCH_OBJECT
def modify_s(self, dn, modlist):
"""Modify the object at dn using the attribute list.
:param dn: an LDAP DN
:param modlist: a list of tuples in the following form:
([MOD_ADD | MOD_DELETE | MOD_REPACE], attribute, value)
if server_fail:
raise ldap.SERVER_DOWN
key = self.key(dn)
LOG.debug('modify item: dn=%(dn)s attrs=%(attrs)s', {
'dn': dn, 'attrs': modlist})
entry = self.db[key]
except KeyError:
LOG.debug('modify item failed: dn=%s not found.', dn)
raise ldap.NO_SUCH_OBJECT
for cmd, k, v in modlist:
values = entry.setdefault(k, [])
if cmd == ldap.MOD_ADD:
v = _internal_attr(k, v)
for x in v:
if x in values:
values += v
elif cmd == ldap.MOD_REPLACE:
values[:] = _internal_attr(k, v)
elif cmd == ldap.MOD_DELETE:
if v is None:
if not values:
LOG.debug('modify item failed: '
'item has no attribute "%s" to delete', k)
values[:] = []
for val in _internal_attr(k, v):
except ValueError:
LOG.debug('modify item failed: '
'item has no attribute "%(k)s" with '
'value "%(v)s" to delete', {
'k': k, 'v': val})
LOG.debug('modify item failed: unknown command %s', cmd)
raise NotImplementedError('modify_s action %s not'
' implemented' % cmd)
self.db[key] = entry
def search_s(self, base, scope,
filterstr='(objectClass=*)', attrlist=None, attrsonly=0):
"""Search for all matching objects under base using the query.
base -- dn to search under
scope -- search scope (base, subtree, onelevel)
filterstr -- filter objects by
attrlist -- attrs to return. Returns all attrs if not specified
if server_fail:
raise ldap.SERVER_DOWN
if (not filterstr) and (scope != ldap.SCOPE_BASE):
raise AssertionError('Search without filter on onelevel or '
'subtree scope')
if scope == ldap.SCOPE_BASE:
item_dict = self.db[self.key(base)]
except KeyError:
LOG.debug('search fail: dn not found for SCOPE_BASE')
raise ldap.NO_SUCH_OBJECT
results = [(base, item_dict)]
elif scope == ldap.SCOPE_SUBTREE:
# FIXME - LDAP search with SUBTREE scope must return the base
# entry, but the code below does _not_. Unfortunately, there are
# several tests that depend on this broken behavior, and fail
# when the base entry is returned in the search results. The
# fix is easy here, just initialize results as above for
# the SCOPE_BASE case.
item_dict = self.db[self.key(base)]
except KeyError:
LOG.debug('search fail: dn not found for SCOPE_SUBTREE')
raise ldap.NO_SUCH_OBJECT
results = [(base, item_dict)]
extraresults = [(k[len(self.__prefix):], v)
for k, v in self.db.items()
if re.match('%s.*,%s' %
re.escape(base)), k)]
elif scope == ldap.SCOPE_ONELEVEL:
def get_entries():
base_dn = ldap.dn.str2dn(base)
base_len = len(base_dn)
for k, v in self.db.items():
if not k.startswith(self.__prefix):
k_dn_str = k[len(self.__prefix):]
k_dn = ldap.dn.str2dn(k_dn_str)
if len(k_dn) != base_len + 1:
if k_dn[-base_len:] != base_dn:
yield (k_dn_str, v)
results = list(get_entries())
# openldap client/server raises PROTOCOL_ERROR for unexpected scope
objects = []
for dn, attrs in results:
# filter the objects by filterstr
id_attr, id_val, _ = ldap.dn.str2dn(dn)[0][0]
match_attrs = attrs.copy()
match_attrs[id_attr] = [id_val]
attrs_checked = set()
if not filterstr or _match_query(filterstr,
if (filterstr and
(scope != ldap.SCOPE_BASE) and
('objectclass' not in attrs_checked)):
raise AssertionError('No objectClass in search filter')
# filter the attributes by attrlist
attrs = {k: v for k, v in attrs.items()
if not attrlist or k in attrlist}
objects.append((dn, attrs))
return objects
def set_option(self, option, invalue):
self._ldap_options[option] = invalue
def get_option(self, option):
value = self._ldap_options.get(option)
return value
def search_ext(self, base, scope,
filterstr='(objectClass=*)', attrlist=None, attrsonly=0,
serverctrls=None, clientctrls=None,
timeout=-1, sizelimit=0):
if clientctrls is not None or timeout != -1 or sizelimit != 0:
raise exception.NotImplemented()
# only passing a single server control is supported by this fake ldap
if len(serverctrls) > 1:
raise exception.NotImplemented()
# search_ext is async and returns an identifier used for
# retrieving the results via result3(). This will be emulated by
# storing the request in a variable with random integer key and
# performing the real lookup in result3()
msgid = random.randint(0, 1000)
PendingRequests[msgid] = (base, scope, filterstr, attrlist, attrsonly,
return msgid
def result3(self, msgid=ldap.RES_ANY, all=1, timeout=None,
"""Execute async request.
Only msgid param is supported. Request info is fetched from global
variable `PendingRequests` by msgid, executed using search_s and
limited if requested.
if all != 1 or timeout is not None or resp_ctrl_classes is not None:
raise exception.NotImplemented()
params = PendingRequests[msgid]
# search_s accepts a subset of parameters of search_ext,
# that's why we use only the first 5.
results = self.search_s(*params[:5])
# extract limit from serverctrl
serverctrls = params[5]
ctrl = serverctrls[0]
if ctrl.size:
rdata = results[:ctrl.size]
rdata = results
# real result3 returns various service info -- rtype, rmsgid,
# serverctrls. Now this info is not used, so all this info is None
rtype = None
rmsgid = None
serverctrls = None
return (rtype, rdata, rmsgid, serverctrls)
class FakeLdapPool(FakeLdap):
"""Emulate the python-ldap API with pooled connections.
This class is used as connector class in PooledLDAPHandler.
def __init__(self, uri, retry_max=None, retry_delay=None, conn=None):
super(FakeLdapPool, self).__init__(conn=conn)
self.url = uri
self.connected = None
self.conn = self
self._connection_time = 5 # any number greater than 0
def get_lifetime(self):
return self._connection_time
def simple_bind_s(self, who=None, cred=None,
serverctrls=None, clientctrls=None):
if self.url.startswith('fakepool://memory'):
if self.url not in FakeShelves:
FakeShelves[self.url] = FakeShelve()
self.db = FakeShelves[self.url]
self.db =[11:])
if not who:
who = 'cn=Admin'
if not cred:
cred = 'password'
super(FakeLdapPool, self).simple_bind_s(who=who, cred=cred,
def unbind_ext_s(self):
"""Added to extend FakeLdap as connector class."""
class FakeLdapNoSubtreeDelete(FakeLdap):
"""FakeLdap subclass that does not support subtree delete.
Same as FakeLdap except delete will throw the LDAP error
ldap.NOT_ALLOWED_ON_NONLEAF if there is an attempt to delete
an entry that has children.
def delete_ext_s(self, dn, serverctrls, clientctrls=None):
"""Remove the ldap object at specified dn."""
if server_fail:
raise ldap.SERVER_DOWN
children = self._getChildren(dn)
if children:
except KeyError:
LOG.debug('delete item failed: dn=%s not found.', dn)
raise ldap.NO_SUCH_OBJECT
super(FakeLdapNoSubtreeDelete, self).delete_ext_s(dn,