codekingpro/portable-devtools
114k
1"""
2"""
3
4# Created on 2015.08.19
5#
6# Author: Giovanni Cannata
7#
8# Copyright 2015 - 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 pyasn1 import __version__ as pyasn1_version
27from pyasn1.codec.ber import decoder # for usage in other modules
28from pyasn1.codec.ber.encoder import Encoder # for monkeypatching of boolean value
29from ..core.results import RESULT_CODES
30from ..utils.conv import to_unicode
31from ..protocol.convert import referrals_to_list
32
33CLASSES = {(False, False): 0, # Universal
34 (False, True): 1, # Application
35 (True, False): 2, # Context
36 (True, True): 3} # Private
37
38
39# Monkeypatching of pyasn1 for encoding Boolean with the value 0xFF for TRUE
40# THIS IS NOT PART OF THE FAST BER DECODER
41if pyasn1_version == 'xxx0.2.3':
42 from pyasn1.codec.ber.encoder import tagMap, BooleanEncoder, encode
43 from pyasn1.type.univ import Boolean
44 from pyasn1.compat.octets import ints2octs
45 class BooleanCEREncoder(BooleanEncoder):
46 _true = ints2octs((255,))
47
48 tagMap[Boolean.tagSet] = BooleanCEREncoder()
49else:
50 from pyasn1.codec.ber.encoder import tagMap, typeMap, AbstractItemEncoder
51 from pyasn1.type.univ import Boolean
52 from copy import deepcopy
53
54 class LDAPBooleanEncoder(AbstractItemEncoder):
55 supportIndefLenMode = False
56 if pyasn1_version <= '0.2.3':
57 from pyasn1.compat.octets import ints2octs
58 _true = ints2octs((255,))
59 _false = ints2octs((0,))
60 def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
61 return value and self._true or self._false, 0
62 elif pyasn1_version <= '0.3.1':
63 def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
64 return value and (255,) or (0,), False, False
65 elif pyasn1_version <= '0.3.4':
66 def encodeValue(self, encodeFun, value, defMode, maxChunkSize, ifNotEmpty=False):
67 return value and (255,) or (0,), False, False
68 elif pyasn1_version <= '0.3.7':
69 def encodeValue(self, value, encodeFun, **options):
70 return value and (255,) or (0,), False, False
71 else:
72 def encodeValue(self, value, asn1Spec, encodeFun, **options):
73 return value and (255,) or (0,), False, False
74
75 customTagMap = deepcopy(tagMap)
76 customTypeMap = deepcopy(typeMap)
77 customTagMap[Boolean.tagSet] = LDAPBooleanEncoder()
78 customTypeMap[Boolean.typeId] = LDAPBooleanEncoder()
79
80 encode = Encoder(customTagMap, customTypeMap)
81# end of monkey patching
82
83# a fast BER decoder for LDAP responses only
84def compute_ber_size(data):
85 """
86 Compute size according to BER definite length rules
87 Returns size of value and value offset
88 """
89
90 if data[1] <= 127: # BER definite length - short form. Highest bit of byte 1 is 0, message length is in the last 7 bits - Value can be up to 127 bytes long
91 return data[1], 2
92 else: # BER definite length - long form. Highest bit of byte 1 is 1, last 7 bits counts the number of following octets containing the value length
93 bytes_length = data[1] - 128
94 value_length = 0
95 cont = bytes_length
96 for byte in data[2: 2 + bytes_length]:
97 cont -= 1
98 value_length += byte * (256 ** cont)
99 return value_length, bytes_length + 2
100
101
102def decode_message_fast(message):
103 ber_len, ber_value_offset = compute_ber_size(get_bytes(message[:10])) # get start of sequence, at maximum 3 bytes for length
104 decoded = decode_sequence(message, ber_value_offset, ber_len + ber_value_offset, LDAP_MESSAGE_CONTEXT)
105 return {
106 'messageID': decoded[0][3],
107 'protocolOp': decoded[1][2],
108 'payload': decoded[1][3],
109 'controls': decoded[2][3] if len(decoded) == 3 else None
110 }
111
112
113def decode_sequence(message, start, stop, context_decoders=None):
114 decoded = []
115 while start < stop:
116 octet = get_byte(message[start])
117 ber_class = CLASSES[(bool(octet & 0b10000000), bool(octet & 0b01000000))]
118 ber_constructed = bool(octet & 0b00100000)
119 ber_type = octet & 0b00011111
120 ber_decoder = DECODERS[(ber_class, octet & 0b00011111)] if ber_class < 2 else None
121 ber_len, ber_value_offset = compute_ber_size(get_bytes(message[start: start + 10]))
122 start += ber_value_offset
123 if ber_decoder:
124 value = ber_decoder(message, start, start + ber_len, context_decoders) # call value decode function
125 else:
126 # try:
127 value = context_decoders[ber_type](message, start, start + ber_len) # call value decode function for context class
128 # except KeyError:
129 # if ber_type == 3: # Referral in result
130 # value = decode_sequence(message, start, start + ber_len)
131 # else:
132 # raise # re-raise, should never happen
133 decoded.append((ber_class, ber_constructed, ber_type, value))
134 start += ber_len
135
136 return decoded
137
138
139def decode_integer(message, start, stop, context_decoders=None):
140 first = message[start]
141 value = -1 if get_byte(first) & 0x80 else 0
142 for octet in message[start: stop]:
143 value = value << 8 | get_byte(octet)
144
145 return value
146
147
148def decode_octet_string(message, start, stop, context_decoders=None):
149 return message[start: stop]
150
151
152def decode_boolean(message, start, stop, context_decoders=None):
153 return False if message[start: stop] == 0 else True
154
155
156def decode_bind_response(message, start, stop, context_decoders=None):
157 return decode_sequence(message, start, stop, BIND_RESPONSE_CONTEXT)
158
159
160def decode_extended_response(message, start, stop, context_decoders=None):
161 return decode_sequence(message, start, stop, EXTENDED_RESPONSE_CONTEXT)
162
163
164def decode_intermediate_response(message, start, stop, context_decoders=None):
165 return decode_sequence(message, start, stop, INTERMEDIATE_RESPONSE_CONTEXT)
166
167
168def decode_controls(message, start, stop, context_decoders=None):
169 return decode_sequence(message, start, stop, CONTROLS_CONTEXT)
170
171
172def ldap_result_to_dict_fast(response):
173 response_dict = dict()
174 response_dict['result'] = int(response[0][3]) # resultCode
175 response_dict['description'] = RESULT_CODES[response_dict['result']]
176 response_dict['dn'] = to_unicode(response[1][3], from_server=True) # matchedDN
177 response_dict['message'] = to_unicode(response[2][3], from_server=True) # diagnosticMessage
178 if len(response) == 4:
179 response_dict['referrals'] = referrals_to_list([to_unicode(referral[3], from_server=True) for referral in response[3][3]]) # referrals
180 else:
181 response_dict['referrals'] = None
182
183 return response_dict
184
185
186######
187
188if str is not bytes: # Python 3
189 def get_byte(x):
190 return x
191
192 def get_bytes(x):
193 return x
194else: # Python 2
195 def get_byte(x):
196 return ord(x)
197
198 def get_bytes(x):
199 return bytearray(x)
200
201DECODERS = {
202 # Universal
203 (0, 1): decode_boolean, # Boolean
204 (0, 2): decode_integer, # Integer
205 (0, 4): decode_octet_string, # Octet String
206 (0, 10): decode_integer, # Enumerated
207 (0, 16): decode_sequence, # Sequence
208 (0, 17): decode_sequence, # Set
209 # Application
210 (1, 1): decode_bind_response, # Bind response
211 (1, 4): decode_sequence, # Search result entry
212 (1, 5): decode_sequence, # Search result done
213 (1, 7): decode_sequence, # Modify response
214 (1, 9): decode_sequence, # Add response
215 (1, 11): decode_sequence, # Delete response
216 (1, 13): decode_sequence, # ModifyDN response
217 (1, 15): decode_sequence, # Compare response
218 (1, 19): decode_sequence, # Search result reference
219 (1, 24): decode_extended_response, # Extended response
220 (1, 25): decode_intermediate_response, # intermediate response
221 (2, 3): decode_octet_string #
222}
223
224BIND_RESPONSE_CONTEXT = {
225 7: decode_octet_string # SaslCredentials
226}
227
228EXTENDED_RESPONSE_CONTEXT = {
229 10: decode_octet_string, # ResponseName
230 11: decode_octet_string # Response Value
231}
232
233INTERMEDIATE_RESPONSE_CONTEXT = {
234 0: decode_octet_string, # IntermediateResponseName
235 1: decode_octet_string # IntermediateResponseValue
236}
237
238LDAP_MESSAGE_CONTEXT = {
239 0: decode_controls, # Controls
240 3: decode_sequence # Referral
241}
242
243CONTROLS_CONTEXT = {
244 0: decode_sequence # Control
245}
246