codekingpro/portable-devtools

1"""
2"""
3
4# Created on 2014.02.02
5#
6# Author: Giovanni Cannata
7#
8# Copyright 2014 - 2020 Giovanni Cannata
9#
10# This file is part of ldap3.
11#
12# ldap3 is free software: you can redistribute it and/or modify
13# it under the terms of the GNU Lesser General Public License as published
14# by the Free Software Foundation, either version 3 of the License, or
15# (at your option) any later version.
16#
17# ldap3 is distributed in the hope that it will be useful,
18# but WITHOUT ANY WARRANTY; without even the implied warranty of
19# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20# GNU Lesser General Public License for more details.
21#
22# You should have received a copy of the GNU Lesser General Public License
23# along with ldap3 in the COPYING and COPYING.LESSER files.
24# If not, see <http://www.gnu.org/licenses/>.
25
26from os import linesep
27
28from .attrDef import AttrDef
29from ..core.exceptions import LDAPKeyError, LDAPObjectError, LDAPAttributeError, LDAPSchemaError
30from .. import STRING_TYPES, SEQUENCE_TYPES, Server, Connection
31from ..protocol.rfc4512 import SchemaInfo, constant_to_class_kind
32from ..protocol.formatters.standard import find_attribute_validator
33from ..utils.ciDict import CaseInsensitiveWithAliasDict
34from ..utils.config import get_config_parameter
35from ..utils.log import log, log_enabled, ERROR, BASIC, PROTOCOL, EXTENDED
36
37
38class ObjectDef(object):
39    """Represent an object in the LDAP server. AttrDefs are stored in a dictionary; the key is the friendly name defined in AttrDef.
40
41    AttrDefs can be added and removed using the += and -= operators
42
43    ObjectDef can be accessed either as a sequence and a dictionary. When accessed the whole AttrDef instance is returned
44
45    """
46    def __init__(self, object_class=None, schema=None, custom_validator=None, auxiliary_class=None):
47        if object_class is None:
48            object_class = []
49
50        if not isinstance(object_class, SEQUENCE_TYPES):
51            object_class = [object_class]
52
53        if auxiliary_class is None:
54            auxiliary_class = []
55
56        if not isinstance(auxiliary_class, SEQUENCE_TYPES):
57            auxiliary_class = [auxiliary_class]
58
59        self.__dict__['_attributes'] = CaseInsensitiveWithAliasDict()
60        self.__dict__['_custom_validator'] = custom_validator
61        self.__dict__['_oid_info'] = []
62
63        if isinstance(schema, Connection) and (schema._deferred_bind or schema._deferred_open):  # probably a lazy connection, tries to bind
64            schema._fire_deferred()
65
66        if schema is not None:
67            if isinstance(schema, Server):
68                schema = schema.schema
69            elif isinstance(schema, Connection):
70                schema = schema.server.schema
71            elif isinstance(schema, SchemaInfo):
72                pass
73            elif schema:
74                error_message = 'unable to read schema'
75                if log_enabled(ERROR):
76                    log(ERROR, '%s for <%s>', error_message, self)
77                raise LDAPSchemaError(error_message)
78            if schema is None:
79                error_message = 'schema not present'
80                if log_enabled(ERROR):
81                    log(ERROR, '%s for <%s>', error_message, self)
82                raise LDAPSchemaError(error_message)
83        self.__dict__['_schema'] = schema
84
85        if self._schema:
86            object_class = [schema.object_classes[name].name[0] for name in object_class]  # uses object class names capitalized as in schema
87            auxiliary_class = [schema.object_classes[name].name[0] for name in auxiliary_class]
88            for object_name in object_class:
89                if object_name:
90                    self._populate_attr_defs(object_name)
91
92            for object_name in auxiliary_class:
93                if object_name:
94                    self._populate_attr_defs(object_name)
95
96        self.__dict__['_object_class'] = object_class
97        self.__dict__['_auxiliary_class'] = auxiliary_class
98
99        if log_enabled(BASIC):
100            log(BASIC, 'instantiated ObjectDef: <%r>', self)
101
102    def _populate_attr_defs(self, object_name):
103        if object_name in self._schema.object_classes:
104            object_schema = self._schema.object_classes[object_name]
105            self.__dict__['_oid_info'].append(object_name + " (" + constant_to_class_kind(object_schema.kind) + ") " + str(object_schema.oid))
106
107            if object_schema.superior:
108                for sup in object_schema.superior:
109                    self._populate_attr_defs(sup)
110            for attribute_name in object_schema.must_contain:
111                self.add_from_schema(attribute_name, True)
112            for attribute_name in object_schema.may_contain:
113                if attribute_name not in self._attributes:  # the attribute could already be defined as "mandatory" in a superclass
114                    self.add_from_schema(attribute_name, False)
115        else:
116            error_message = 'object class \'%s\' not defined in schema' % object_name
117            if log_enabled(ERROR):
118                log(ERROR, '%s for <%s>', error_message, self)
119            raise LDAPObjectError(error_message)
120
121    def __repr__(self):
122        if self._object_class:
123            r = 'OBJ : ' + ', '.join(self._object_class) + linesep
124        else:
125            r = 'OBJ : <None>' + linesep
126        if self._auxiliary_class:
127            r += 'AUX : ' + ', '.join(self._auxiliary_class) + linesep
128        else:
129            r += 'AUX : <None>' + linesep
130        r += 'OID: ' + ', '.join([oid for oid in self._oid_info]) + linesep
131        r += 'MUST: ' + ', '.join(sorted([attr for attr in self._attributes if self._attributes[attr].mandatory])) + linesep
132        r += 'MAY : ' + ', '.join(sorted([attr for attr in self._attributes if not self._attributes[attr].mandatory])) + linesep
133
134        return r
135
136    def __str__(self):
137        return self.__repr__()
138
139    def __getitem__(self, item):
140        return self.__getattr__(item)
141
142    def __getattr__(self, item):
143        item = ''.join(item.split()).lower()
144        if '_attributes' in self.__dict__:
145            try:
146                return self._attributes[item]
147            except KeyError:
148                error_message = 'key \'%s\' not present' % item
149                if log_enabled(ERROR):
150                    log(ERROR, '%s for <%s>', error_message, self)
151                raise LDAPKeyError(error_message)
152        else:
153            error_message = 'internal _attributes property not defined'
154            if log_enabled(ERROR):
155                log(ERROR, '%s for <%s>', error_message, self)
156            raise LDAPKeyError(error_message)
157
158    def __setattr__(self, key, value):
159        error_message = 'object \'%s\' is read only' % key
160        if log_enabled(ERROR):
161            log(ERROR, '%s for <%s>', error_message, self)
162        raise LDAPObjectError(error_message)
163
164    def __iadd__(self, other):
165        self.add_attribute(other)
166        return self
167
168    def __isub__(self, other):
169        if isinstance(other, AttrDef):
170            self.remove_attribute(other.key)
171        elif isinstance(other, STRING_TYPES):
172            self.remove_attribute(other)
173
174        return self
175
176    def __iter__(self):
177        for attribute in self._attributes:
178            yield self._attributes[attribute]
179
180    def __len__(self):
181        return len(self._attributes)
182
183    if str is not bytes:  # Python 3
184        def __bool__(self):  # needed to make the objectDef appears as existing in "if cursor:" even if there are no entries
185            return True
186    else:  # Python 2
187        def __nonzero__(self):
188            return True
189
190    def __contains__(self, item):
191        try:
192            self.__getitem__(item)
193        except KeyError:
194            return False
195
196        return True
197
198    def add_from_schema(self, attribute_name, mandatory=False):
199            attr_def = AttrDef(attribute_name)
200            attr_def.validate = find_attribute_validator(self._schema, attribute_name, self._custom_validator)
201            attr_def.mandatory = mandatory  # in schema mandatory is specified in the object class, not in the attribute class
202            if self._schema and self._schema.attribute_types and attribute_name in self._schema.attribute_types:
203                attr_def.single_value = self._schema.attribute_types[attribute_name].single_value
204                attr_def.oid_info = self._schema.attribute_types[attribute_name]
205            self.add_attribute(attr_def)
206
207    def add_attribute(self, definition=None):
208        """Add an AttrDef to the ObjectDef. Can be called with the += operator.
209        :param definition: the AttrDef object to add, can also be a string containing the name of attribute to add. Can be a list of both
210
211        """
212        conf_attributes_excluded_from_object_def = [v.lower() for v in get_config_parameter('ATTRIBUTES_EXCLUDED_FROM_OBJECT_DEF')]
213        if isinstance(definition, STRING_TYPES):
214            self.add_from_schema(definition)
215        elif isinstance(definition, AttrDef):
216            if definition.key.lower() not in conf_attributes_excluded_from_object_def:
217                if definition.key not in self._attributes:
218                    self._attributes[definition.key] = definition
219                    if definition.name and definition.name != definition.key:
220                        self._attributes.set_alias(definition.key, definition.name)
221                    other_names = [name for name in definition.oid_info.name if definition.key.lower() != name.lower()] if definition.oid_info else None
222                    if other_names:
223                        self._attributes.set_alias(definition.key, other_names)
224
225                if not definition.validate:
226                    validator = find_attribute_validator(self._schema, definition.key, self._custom_validator)
227                    self._attributes[definition.key].validate = validator
228        elif isinstance(definition, SEQUENCE_TYPES):
229            for element in definition:
230                self.add_attribute(element)
231        else:
232            error_message = 'unable to add element to object definition'
233            if log_enabled(ERROR):
234                log(ERROR, '%s for <%s>', error_message, self)
235            raise LDAPObjectError(error_message)
236
237    def remove_attribute(self, item):
238        """Remove an AttrDef from the ObjectDef. Can be called with the -= operator.
239        :param item: the AttrDef to remove, can also be a string containing the name of attribute to remove
240
241        """
242        key = None
243        if isinstance(item, STRING_TYPES):
244            key = ''.join(item.split()).lower()
245        elif isinstance(item, AttrDef):
246            key = item.key.lower()
247
248        if key:
249            for attr in self._attributes:
250                if key == attr.lower():
251                    del self._attributes[attr]
252                    break
253            else:
254                error_message = 'key \'%s\' not present' % key
255                if log_enabled(ERROR):
256                    log(ERROR, '%s for <%s>', error_message, self)
257                raise LDAPKeyError(error_message)
258        else:
259            error_message = 'key type must be str or AttrDef not ' + str(type(item))
260            if log_enabled(ERROR):
261                log(ERROR, '%s for <%s>', error_message, self)
262            raise LDAPAttributeError(error_message)
263
264    def clear_attributes(self):
265        """Empty the ObjectDef attribute list
266
267        """
268        self.__dict__['object_class'] = None
269        self.__dict__['auxiliary_class'] = None
270        self.__dict__['_attributes'] = dict()
271