1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
"""
"""
# Created on 2013.07.15
#
# Author: Giovanni Cannata
#
# Copyright 2015 Giovanni Cannata
#
# This file is part of ldap3.
#
# ldap3 is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# ldap3 is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with ldap3 in the COPYING and COPYING.LESSER files.
# If not, see <http://www.gnu.org/licenses/>.
from threading import Thread, Lock
import socket
from .. import RESPONSE_COMPLETE, get_config_parameter
from ..core.exceptions import LDAPSSLConfigurationError, LDAPStartTLSError, LDAPOperationResult
from ..strategy.base import BaseStrategy
from ..protocol.rfc4511 import LDAPMessage
from ..utils.log import log, log_enabled, format_ldap_message, ERROR, NETWORK, EXTENDED
from ..utils.asn1 import decoder, decode_message_fast
# noinspection PyProtectedMember
class AsyncStrategy(BaseStrategy):
"""
This strategy is asynchronous. You send the request and get the messageId of the request sent
Receiving data from socket is managed in a separated thread in a blocking mode
Requests return an int value to indicate the messageId of the requested Operation
You get the response with get_response, it has a timeout to wait for response to appear
Connection.response will contain the whole LDAP response for the messageId requested in a dict form
Connection.request will contain the result LDAP message in a dict form
Response appear in strategy._responses dictionary
"""
# noinspection PyProtectedMember
class ReceiverSocketThread(Thread):
"""
The thread that actually manage the receiver socket
"""
def __init__(self, ldap_connection):
Thread.__init__(self)
self.connection = ldap_connection
def run(self):
"""
Wait for data on socket, compute the length of the message and wait for enough bytes to decode the message
Message are appended to strategy._responses
"""
socket_size = get_config_parameter('SOCKET_SIZE')
unprocessed = b''
get_more_data = True
listen = True
data = b''
while listen:
if get_more_data:
try:
data = self.connection.socket.recv(socket_size)
except (OSError, socket.error):
listen = False
except Exception as e:
if log_enabled(ERROR):
log(ERROR, '<%s> for <%s>', str(e), self.connection)
raise # unexpected exception - re-raise
if len(data) > 0:
unprocessed += data
data = b''
else:
listen = False
length = BaseStrategy.compute_ldap_message_size(unprocessed)
if length == -1 or len(unprocessed) < length:
get_more_data = True
elif len(unprocessed) >= length: # add message to message list
if self.connection.usage:
self.connection._usage.update_received_message(length)
if log_enabled(NETWORK):
log(NETWORK, 'received %d bytes via <%s>', length, self.connection)
if self.connection.fast_decoder:
ldap_resp = decode_message_fast(unprocessed[:length])
dict_response = self.connection.strategy.decode_response_fast(ldap_resp)
else:
ldap_resp = decoder.decode(unprocessed[:length], asn1Spec=LDAPMessage())[0]
dict_response = self.connection.strategy.decode_response(ldap_resp)
message_id = int(ldap_resp['messageID'])
if log_enabled(NETWORK):
log(NETWORK, 'received 1 ldap message via <%s>', self.connection)
if log_enabled(EXTENDED):
log(EXTENDED, 'ldap message received via <%s>:%s', self.connection, format_ldap_message(ldap_resp, '<<'))
if dict_response['type'] == 'extendedResp' and dict_response['responseName'] == '1.3.6.1.4.1.1466.20037':
if dict_response['result'] == 0: # StartTls in progress
if self.connection.server.tls:
self.connection.server.tls._start_tls(self.connection)
else:
self.connection.last_error = 'no Tls object defined in Server'
if log_enabled(ERROR):
log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
raise LDAPSSLConfigurationError(self.connection.last_error)
else:
self.connection.last_error = 'asynchronous StartTls failed'
if log_enabled(ERROR):
log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
raise LDAPStartTLSError(self.connection.last_error)
if message_id != 0: # 0 is reserved for 'Unsolicited Notification' from server as per RFC4511 (paragraph 4.4)
with self.connection.strategy.lock:
if message_id in self.connection.strategy._responses:
self.connection.strategy._responses[message_id].append(dict_response)
else:
self.connection.strategy._responses[message_id] = [dict_response]
if dict_response['type'] not in ['searchResEntry', 'searchResRef', 'intermediateResponse']:
self.connection.strategy._responses[message_id].append(RESPONSE_COMPLETE)
unprocessed = unprocessed[length:]
get_more_data = False if unprocessed else True
listen = True if self.connection.listening or unprocessed else False
else: # Unsolicited Notification
if dict_response['responseName'] == '1.3.6.1.4.1.1466.20036': # Notice of Disconnection as per RFC4511 (paragraph 4.4.1)
listen = False
else:
self.connection.last_error = 'unknown unsolicited notification from server'
if log_enabled(ERROR):
log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
raise LDAPStartTLSError(self.connection.last_error)
self.connection.strategy.close()
def __init__(self, ldap_connection):
BaseStrategy.__init__(self, ldap_connection)
self.sync = False
self.no_real_dsa = False
self.pooled = False
self._responses = None
self.can_stream = False
self.receiver = None
self.lock = Lock()
def open(self, reset_usage=True, read_server_info=True):
"""
Open connection and start listen on the socket in a different thread
"""
with self.lock:
self._responses = dict()
BaseStrategy.open(self, reset_usage, read_server_info)
if read_server_info:
try:
self.connection.refresh_server_info()
except LDAPOperationResult: # catch errors from server if raise_exception = True
self.connection.server._dsa_info = None
self.connection.server._schema_info = None
def close(self):
"""
Close connection and stop socket thread
"""
with self.lock:
BaseStrategy.close(self)
def post_send_search(self, message_id):
"""
Clears connection.response and returns messageId
"""
self.connection.response = None
self.connection.result = message_id
return message_id
def post_send_single_response(self, message_id):
"""
Clears connection.response and returns messageId.
"""
self.connection.response = None
self.connection.result = message_id
return message_id
def _start_listen(self):
"""
Start thread in daemon mode
"""
if not self.connection.listening:
self.receiver = AsyncStrategy.ReceiverSocketThread(self.connection)
self.connection.listening = True
self.receiver.daemon = True
self.receiver.start()
def _get_response(self, message_id):
"""
Performs the capture of LDAP response for this strategy
Checks lock to avoid race condition with receiver thread
"""
with self.lock:
responses = self._responses.pop(message_id) if message_id in self._responses and self._responses[message_id][-1] == RESPONSE_COMPLETE else None
return responses
def receiving(self):
raise NotImplementedError
def get_stream(self):
raise NotImplementedError
def set_stream(self, value):
raise NotImplementedError
|
