fakeldap.py

# coding: utf-8

# Copyright (c) 2009, Peter Sagerson
# Copyright (c) 2011, Christo Buschek <crito@30loops.net>
# Copyright (c) 2013, Roberto Polli <rpolli@babel.it>
# All rights reserved.
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 
# - Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
# 
# - Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import sys
import logging
import types
import ldap
from collections import defaultdict


__version__ = "0.5.1"



def asynchronous(wrapped_f):
    """Decorator for supporting async functions like search, modify,... """
    import uuid
    def f(self, *args, **kwds):
        item = wrapped_f(self, *args, **kwds)
        item_uid = str(uuid.uuid4())
        try:
            self.returned[item_uid] = item
        except AttributeError:
            self.returned = {item_uid: item}

        #print "stored item %s (%r) = %r %r " % (wrapped_f.__name__, [self, args, kwds], item_uid, item)
        return item_uid
    # preserve function name. Consider using the itertools.wrap decorator 
    f.__name__ = wrapped_f.__name__
    return f


def leq(entry_attributes, needle):
    """Attribute equality or contain"""
    #print("leq: %r in %r" % (needle, entry_attributes))
    return (str(needle) == str(entry_attributes) 
        or str(needle) in map(str, entry_attributes))
    
def lneq(x,y):
    return not leq(x,y)
    
def land(*args):
    #print("land: %r" % map(str,args))
    return all(args)
    
def lor(*args):
    return any(args)
    
def evaluate(expression, d):
    """Evaluate an expression generated by 
        fakeldap.clause function.
        @param expression - an expression
        @param d - a directory entry like dict
    """
    # an expression is either:
    # * a boolean (already evaluated)
    # or a triple: (operator, expr or string, expr or string)
    if isinstance(expression, bool):
        return expression
     
    try:
        op = expression[0]
        if isinstance(expression[1], str):
            # in this case, even b should be
            # str or int. 
            a, b = expression[1:3]
            try:
                return op(d[a], b)
            except KeyError:
                return False
        # pre-evaluate expression members
        args = [evaluate(j, d) for j in expression[1:]]
        return op(*args)
    except ValueError, e:
        raise ValueError("Unexpected Expression: %r" % expression, e)
        
def clause(s):
    """Yields an expression to be evaluated"""
    assert s[0]+s[-1] == "()", "Malformed clause: %s" % s
    s = s[1:-1]
    ret = []
    if s[0] not in "&|":
        print("simple hit: %r" % s)
        # simple parse :)
        x, y = s.split("=", 1)
        return [leq, x, y]
    else:
        print("mixed hit %r" % s)
        ret.append({'&': land, '|': lor}[s[0]])
        #remove operator from s
        for x in _subexpressions(s[1:]):
            ret.append(clause(x))
        return ret

def _subexpressions(s):
    """Takes the 1-level subexpressions of a filter
        (o=1) -> (o=1)
        (o=1)(o=2) -> (o=1), (o=2)
        (|(o=1)(o=2))(o=3) -> (|(o=1)(o=2)), (o=3)
    
    """
    l, i, start = 0, 0, 0
    for x in s:
        if x=='(': 
            l+=1
        if x==')': 
            l-=1
            if l == 0:
                print("subexpression %r" % s[start:i+1])
                yield s[start:i+1]
                start=i+1
        i += 1
        
        

class MockLDAP: # here I had to remove the new-style class definition because SimpleLDAPObject uses old-style classes
    """
    This is a stand-in for the python-ldap module; it serves as both the ldap
    module and the LDAPObject class. While it's temping to add some real LDAP
    capabilities here, this is designed to remain as simple as possible, so as
    to minimize the risk of creating bogus unit tests through a buggy test
    harness.

    Simple operations can be simulated, but for nontrivial searches, the client
    will have to seed the mock object with return values for expected API calls.
    This may sound like cheating, but it's really no more so than a simulated
    LDAP server. The fact is we can not require python-ldap to be installed in
    order to run the unit tests, so all we can do is verify that LDAPBackend is
    calling the APIs that we expect.

    set_return_value takes the name of an API, a tuple of arguments, and a
    return value. Every time an API is called, it looks for a predetermined
    return value based on the arguments received. If it finds one, then it
    returns it, or raises it if it's an Exception. If it doesn't find one, then
    it tries to satisfy the request internally. If it can't, it raises a
    PresetReturnRequiredError.

    At any time, the client may call ldap_methods_called_with_arguments() or
    ldap_methods_called() to get a record of all of the LDAP API calls that have
    been made, with or without arguments.
    """

    class PresetReturnRequiredError(Exception): pass

    SCOPE_BASE = 0
    SCOPE_ONELEVEL = 1
    SCOPE_SUBTREE = 2

    MOD_ADD = 0
    MOD_DELETE = 1
    MOD_REPLACE = 2

    class LDAPError(Exception): pass
    class INVALID_CREDENTIALS(LDAPError): pass
    class NO_SUCH_OBJECT(LDAPError): pass
    class ALREADY_EXISTS(LDAPError): pass

    #
    # Submodules
    #
    class dn(object):
        def escape_dn_chars(s):
            return s
        escape_dn_chars = staticmethod(escape_dn_chars)

    class filter(object):
        def escape_filter_chars(s):
            return s
        escape_filter_chars = staticmethod(escape_filter_chars)


    def __init__(self, uri=None, directory=None, *args, **kwds): # SimpleLDAPObject takes uri as the first parameter!
        """
        directory is a complex structure with the entire contents of the
        mock LDAP directory. directory must be a dictionary mapping
        distinguished names to dictionaries of attributes. Each attribute
        dictionary maps attribute names to lists of values. e.g.:

        {
            "uid=alice,ou=users,dc=example,dc=com":
            {
                "uid": ["alice"],
                "userPassword": ["secret"],
            },
        }
        """
        if directory is not None:
            self.directory = directory
        else:
            self.directory = defaultdict(lambda: {})

        self.reset()

    def reset(self):
        """
        Resets our recorded API calls and queued return values as well as
        miscellaneous configuration options.
        """
        self.calls = []
        self.return_value_maps = defaultdict(lambda: {})
        self.options = {}
        self.tls_enabled = False

    def set_return_value(self, api_name, arguments, value):
        """
        Stores a preset return value for a given API with a given set of
        arguments.
        """
        # hack, cause lists are not hashable
        if type(arguments[1]) is types.ListType:
            arguments[1] = tuple(arguments[1])
        print "Set value. api_name: %s, arguments: %s, value: %s" % (api_name, arguments, value)
        self.return_value_maps[api_name][arguments] = value

    def ldap_methods_called_with_arguments(self):
        """
        Returns a list of 2-tuples, one for each API call made since the last
        reset. Each tuple contains the name of the API and a dictionary of
        arguments. Argument defaults are included.
        """
        return self.calls

    def ldap_methods_called(self):
        """
        Returns the list of API names called.
        """
        return [call[0] for call in self.calls]

    #
    # Begin LDAP methods
    #

    def set_option(self, option, invalue):
        self._record_call('set_option', {
            'option': option,
            'invalue': invalue
        })

        self.options[option] = invalue

    def initialize(self, uri, trace_level=0, trace_file=sys.stdout, trace_stack_limit=None):
        self._record_call('initialize', {
            'uri': uri,
            'trace_level': trace_level,
            'trace_file': trace_file,
            'trace_stack_limit': trace_stack_limit
        })

        value = self._get_return_value('initialize',
            (uri, trace_level, trace_file, trace_stack_limit))
        if value is None:
            value = self

        return value

    def simple_bind_s(self, who='', cred=''):
        self._record_call('simple_bind_s', {
            'who': who,
            'cred': cred
        })

        value = self._get_return_value('simple_bind_s', (who, cred))
        if value is None:
            value = self._simple_bind_s(who, cred)

        return value

    def search_s(self, base, scope, filterstr='(objectClass=*)', attrlist=None, attrsonly=0):
        # Hack, cause attributes as a list can't be hashed for storing it
        if type(attrlist) is types.ListType:
            attrlist = ', '.join(attrlist)

        self._record_call('search_s', {
            'base': base,
            'scope': scope,
            'filterstr':filterstr,
            'attrlist':attrlist,
            'attrsonly':attrsonly
        })
        value = self._get_return_value('search_s',
            (base, scope, filterstr, attrlist, attrsonly))
        if value is None:
            value = self._search_s(base, scope, filterstr, attrlist, attrsonly)

        return value

    @asynchronous
    def search(self, base, scope, filterstr='(objectClass=*)', attrlist=None, attrsonly=0):
        ret = self.search_s(base, scope, filterstr, attrlist, attrsonly)
        if not ret:
            return []
        return ret[0]

    def start_tls_s(self):
        self.tls_enabled = True

    def compare_s(self, dn, attr, value):
        self._record_call('compare_s', {
            'dn': dn,
            'attr': attr,
            'value': value
        })

        result = self._get_return_value('compare_s', (dn, attr, value))
        if result is None:
            result = self._compare_s(dn, attr, value)

        # print "compare_s('%s', '%s', '%s'): %d" % (dn, attr, value, result)

        return result

    def modify_s(self, dn, mod_attrs):
        self._record_call('modify_s', {
            'dn': dn,
            'mod_attrs': mod_attrs
        })

        mod_attrs = tuple(mod_attrs)
        result = self._get_return_value('modify_s', (dn, mod_attrs))
        if result is None:
            result = self._modify_s(dn, mod_attrs)

        return result

    @asynchronous
    def modify(self, dn, mod_attrs):
        """Mock asynchronous modify
        """
        return self.modify_s(dn, mod_attrs)

    def result(self, res):
        """Mock async result retrieval"""
        #print "retrieving %r" % [res]
        try:
            return 999, self.returned.pop(res)
        except KeyError:
            raise MockLDAP.NO_SUCH_OBJECT

    def delete_s(self, dn):
        self._record_call('delete_s', {
            'dn': dn
        })

        result = self._get_return_value('delete_s', dn)
        if result is None:
            result = self._delete_s(dn)

        return result

    def add_s(self, dn, record):
        self._record_call('add_s', {
            'dn': dn,
            'record': record,
        })

        record = self._mangle_record(record)

        result = self._get_return_value('add_s', (dn, record))
        if result is None:
            result = self._add_s(dn, record)

        return result

    def rename_s(self, dn, newdn):
        self._record_call('rename_s', {
            'dn': dn,
            'newdn': newdn,
        })

        result = self._get_return_value('rename_s', (dn, newdn))
        if result is None:
            result = self._rename_s(dn, newdn)

        return result

    #
    # Internal implementations
    #

    def _simple_bind_s(self, who='', cred=''):
        success = False

        if(who == '' and cred == ''):
            success = True
        elif self._compare_s(who.lower(), 'userPassword', cred):
            success = True

        if success:
            return (97, []) # python-ldap returns this; I don't know what it means
        else:
            raise self.INVALID_CREDENTIALS('%s:%s' % (who, cred))

    def _compare_s(self, dn, attr, value):
        try:
            found = (value in self.directory[dn][attr])
        except KeyError:
            found = False

        return found and 1 or 0

    def _modify_s(self, dn, mod_attrs):
        try:
            entry = self.directory[dn]
        except KeyError:
            raise ldap.NO_SUCH_OBJECT

        for item in mod_attrs:
            op, key, value = item
            if op is 0:
                # FIXME: Can't handle multiple entries with the same name
                # its broken right now
                # do a MOD_ADD, assume it to be a list of values
                entry[key] = value
            elif op is 1:
                # do a MOD_DELETE
                row = entry[key]
                if row is types.ListType:
                    for i in range(len(row)):
                        if value is row[i]:
                            del row[i]
                else:
                    del entry[key]
                self.directory[dn] = entry
            elif op is 2:
                # do a MOD_REPLACE
                entry[key] = value

        self.directory[dn] = entry

        return (103, [])

    def _rename_s(self, dn, newdn):
        try:
            entry = self.directory[dn]
        except KeyError:
            raise ldap.NO_SUCH_OBJECT

        changes = newdn.split('=')
        newfulldn = '%s=%s,%s' % (changes[0], changes[1],
                ','.join(dn.split(',')[1:]))

        entry[changes[0]] = changes[1]
        self.directory[newfulldn] = entry
        del self.directory[dn]

        return (109, [])

    def _delete_s(self, dn):
        try:
            del self.directory[dn]
        except KeyError:
            raise ldap.NO_SUCH_OBJECT

        return (107, [])

    def _search_s(self, base, scope, filterstr, attrlist, attrsonly):
        """
        We can do a SCOPE_BASE search with the default filter. Beyond that,
        you're on your own.
        """
        # Subtree implementation
        def subtree_search(base, filterstr):                
            entries_in_base = [x for x in self.directory if x.endswith(base)]
            clauser = clause(filterstr)
            for dn in entries_in_base:
                if evaluate(clauser, self.directory[dn]):
                    yield (dn, self.directory[dn])

        if not base in self.directory:
            # This is the only case in which
            # we should raise NO_SUCH_OBJECT
            # In all the other cases we should
            # just return an empty list
            raise ldap.NO_SUCH_OBJECT
    
        if scope == ldap.SCOPE_SUBTREE:
            return list(subtree_search(base, filterstr))
            
                
        #FIXME: Implement different scopes
#        if scope != self.SCOPE_BASE:
#            raise self.PresetReturnRequiredError('search_s("%s", %d, "%s", "%s", %d)' %
#                (base, scope, filterstr, attrlist, attrsonly))

#        if filterstr != '(objectClass=*)':
#            raise self.PresetReturnRequiredError('search_s("%s", %d, "%s", "%s", %d)' %
#                (base, scope, filterstr, attrlist, attrsonly))

        def _finditem(obj, key):
            if key in obj: 
                # exact match
                return obj[key]
            found = None
            for k, v in obj.items():
                if isinstance(v,dict):
                    found =  _finditem(v, key)  #added return statement
            if found is not None:
                return found
            else:
                return None

        #attrs = self.directory.get(base)
        attrs = _finditem(self.directory, base)
        if attrs is None:
            raise ldap.NO_SUCH_OBJECT

        return [(base, self.filter_attrs(attrlist, attrs))]

    def filter_attrs(self, attrlist, attrs):
        if attrlist:
            return {name: attr for name, attr in attrs.iteritems() if name in attrlist}
        else:
            return attrs

    def _add_s(self, dn, record):
        # change the record into the proper format for the internal directory
        entry = {}
        for item in record:
            entry[item[0]] = item[1]
        try:
            self.directory[dn]
            raise ldap.ALREADY_EXISTS
        except KeyError:
            self.directory[dn] = entry
            return (105,[], len(self.calls), [])

    #
    # Utils
    #

    def _mangle_record(self, record):
        """Change lists into tuples, so that they can be hashed."""
        new_record = []

        for item in record:
            key, value = item
            if type(value) is types.ListType:
                value = tuple(value)
            new_record.append((key, value))

        if type(new_record) is types.ListType:
            new_record = tuple(new_record)

        return new_record

    def _record_call(self, api_name, arguments):
        self.calls.append((api_name, arguments))

    def _get_return_value(self, api_name, arguments):
        try:
            print "api: %s, arguments: %s" % (api_name, arguments)
            value = self.return_value_maps[api_name][arguments]
        except KeyError:
            value = None

        if isinstance(value, Exception):
            raise value

        return value