@@ -1,550 +0,0 @@
1−import atexit
2−import sys
3−import threading
4−import asyncio
5−import logging
6−import concurrent.futures
7−from time import time
8−from itertools import cycle
9−
10−import httpcore
11−import httpx
12−import h2.exceptions
13−from httpx_socks import AsyncProxyTransport
14−from python_socks import parse_proxy_url
15−import python_socks._errors
16−
17−from searx import settings
18−from searx import logger
19−from searx.raise_for_httperror import raise_for_httperror
20−
21−# Optional uvloop (support Python 3.6)
22−try:
23− import uvloop
24−except ImportError:
25− pass
26−else:
27− uvloop.install()
28−
29−# queue.SimpleQueue: Support Python 3.6
30−try:
31− from queue import SimpleQueue
32−except ImportError:
33− from queue import Empty
34− from collections import deque
35−
36− class SimpleQueue:
37− """Minimal backport of queue.SimpleQueue"""
38−
39− def __init__(self):
40− self._queue = deque()
41− self._count = threading.Semaphore(0)
42−
43− def put(self, item):
44− self._queue.append(item)
45− self._count.release()
46−
47− def get(self, timeout=None):
48− if not self._count.acquire(True, timeout):
49− raise Empty
50− return self._queue.popleft()
51−
52−
53−logger = logger.getChild('poolrequests')
54−
55−
56−try:
57− import ssl
58− if ssl.OPENSSL_VERSION_INFO[0:3] < (1, 0, 2):
59− # https://github.com/certifi/python-certifi#1024-bit-root-certificates
60− logger.critical('You are using an old openssl version({0}), please upgrade above 1.0.2!'
61− .format(ssl.OPENSSL_VERSION))
62− sys.exit(1)
63−except ImportError:
64− ssl = None
65−if not getattr(ssl, "HAS_SNI", False):
66− try:
67− import OpenSSL # pylint: disable=unused-import
68− except ImportError:
69− logger.critical("ssl doesn't support SNI and the pyopenssl module is not installed.\n"
70− "Some HTTPS connections will fail")
71− sys.exit(1)
72−
73−
74−LOOP = None
75−CLIENTS = dict()
76−THREADLOCAL = threading.local()
77−LIMITS = httpx.Limits(
78− # Magic number kept from previous code
79− max_connections=settings['outgoing'].get('pool_connections', 100),
80− # Picked from constructor
81− max_keepalive_connections=settings['outgoing'].get('pool_maxsize', 10),
82− #
83− keepalive_expiry=settings['outgoing'].get('keepalive_expiry', 5.0)
84−)
85−# default parameters for AsyncHTTPTransport
86−# see https://github.com/encode/httpx/blob/e05a5372eb6172287458b37447c30f650047e1b8/httpx/_transports/default.py#L108-L121 # noqa
87−TRANSPORT_KWARGS = {
88− 'http2': settings['outgoing'].get('http2', False),
89− 'retries': 0,
90− 'trust_env': False,
91− 'backend': 'asyncio'
92−}
93−# requests compatibility when reading proxy settings from settings.yml
94−PROXY_PATTERN_MAPPING = {
95− 'http': 'https://',
96− 'https:': 'https://'
97−}
98−# default maximum redirect
99−# from https://github.com/psf/requests/blob/8c211a96cdbe9fe320d63d9e1ae15c5c07e179f8/requests/models.py#L55
100−DEFAULT_REDIRECT_LIMIT = 30
101−
102−
103−if settings['outgoing'].get('source_ips'):
104− LOCAL_ADDRESS_CYCLE = cycle(settings['outgoing'].get('source_ips'))
105−else:
106− LOCAL_ADDRESS_CYCLE = cycle((None, ))
107−
108−
109−def set_timeout_for_thread(timeout, start_time=None):
110− THREADLOCAL.timeout = timeout
111− THREADLOCAL.start_time = start_time
112−
113−
114−def set_enable_http_protocol(enable_http):
115− THREADLOCAL.enable_http = enable_http
116−
117−
118−def get_enable_http_protocol():
119− try:
120− return THREADLOCAL.enable_http
121− except AttributeError:
122− return False
123−
124−
125−def reset_time_for_thread():
126− THREADLOCAL.total_time = 0
127−
128−
129−def get_time_for_thread():
130− return THREADLOCAL.total_time
131−
132−
133−def get_proxy_cycles(proxy_settings):
134− if not proxy_settings:
135− return None
136− # Backwards compatibility for single proxy in settings.yml
137− for protocol, proxy in proxy_settings.items():
138− if isinstance(proxy, str):
139− proxy_settings[protocol] = [proxy]
140−
141− for protocol in proxy_settings:
142− proxy_settings[protocol] = cycle(proxy_settings[protocol])
143− return proxy_settings
144−
145−
146−GLOBAL_PROXY_CYCLES = get_proxy_cycles(settings['outgoing'].get('proxies'))
147−
148−
149−def get_proxies(proxy_cycles):
150− if proxy_cycles:
151− return {protocol: next(proxy_cycle) for protocol, proxy_cycle in proxy_cycles.items()}
152− return None
153−
154−
155−def get_global_proxies():
156− return get_proxies(GLOBAL_PROXY_CYCLES)
157−
158−
159−async def close_connections_for_url(connection_pool: httpcore.AsyncConnectionPool, url: httpcore._utils.URL):
160− origin = httpcore._utils.url_to_origin(url)
161− logger.debug('Drop connections for %r', origin)
162− connections_to_close = connection_pool._connections_for_origin(origin)
163− for connection in connections_to_close:
164− await connection_pool._remove_from_pool(connection)
165− try:
166− await connection.aclose()
167− except httpcore.NetworkError as e:
168− logger.warning('Error closing an existing connection', exc_info=e)
169−
170−
171−class AsyncHTTPTransportNoHttp(httpcore.AsyncHTTPTransport):
172− """Block HTTP request"""
173−
174− async def arequest(self, method, url, headers=None, stream=None, ext=None):
175− raise httpcore.UnsupportedProtocol("HTTP protocol is disabled")
176−
177−
178−class AsyncProxyTransportFixed(AsyncProxyTransport):
179− """Fix httpx_socks.AsyncProxyTransport
180−
181− Map python_socks exceptions to httpcore.ProxyError
182−
183− Map socket.gaierror to httpcore.ConnectError
184−
185− Note: keepalive_expiry is ignored, AsyncProxyTransport should call:
186− * self._keepalive_sweep()
187− * self._response_closed(self, connection)
188−
189− Note: AsyncProxyTransport inherit from AsyncConnectionPool
190−
191− Note: the API is going to change on httpx 0.18.0
192− see https://github.com/encode/httpx/pull/1522
193− """
194−
195− async def arequest(self, method, url, headers=None, stream=None, ext=None):
196− retry = 2
197− while retry > 0:
198− retry -= 1
199− try:
200− return await super().arequest(method, url, headers, stream, ext)
201− except (python_socks._errors.ProxyConnectionError,
202− python_socks._errors.ProxyTimeoutError,
203− python_socks._errors.ProxyError) as e:
204− raise httpcore.ProxyError(e)
205− except OSError as e:
206− # socket.gaierror when DNS resolution fails
207− raise httpcore.NetworkError(e)
208− except httpcore.RemoteProtocolError as e:
209− # in case of httpcore.RemoteProtocolError: Server disconnected
210− await close_connections_for_url(self, url)
211− logger.warning('httpcore.RemoteProtocolError: retry', exc_info=e)
212− # retry
213− except (httpcore.NetworkError, httpcore.ProtocolError) as e:
214− # httpcore.WriteError on HTTP/2 connection leaves a new opened stream
215− # then each new request creates a new stream and raise the same WriteError
216− await close_connections_for_url(self, url)
217− raise e
218−
219−
220−class AsyncHTTPTransportFixed(httpx.AsyncHTTPTransport):
221− """Fix httpx.AsyncHTTPTransport"""
222−
223− async def arequest(self, method, url, headers=None, stream=None, ext=None):
224− retry = 2
225− while retry > 0:
226− retry -= 1
227− try:
228− return await super().arequest(method, url, headers, stream, ext)
229− except OSError as e:
230− # socket.gaierror when DNS resolution fails
231− raise httpcore.ConnectError(e)
232− except httpcore.CloseError as e:
233− # httpcore.CloseError: [Errno 104] Connection reset by peer
234− # raised by _keepalive_sweep()
235− # from https://github.com/encode/httpcore/blob/4b662b5c42378a61e54d673b4c949420102379f5/httpcore/_backends/asyncio.py#L198 # noqa
236− await close_connections_for_url(self._pool, url)
237− logger.warning('httpcore.CloseError: retry', exc_info=e)
238− # retry
239− except httpcore.RemoteProtocolError as e:
240− # in case of httpcore.RemoteProtocolError: Server disconnected
241− await close_connections_for_url(self._pool, url)
242− logger.warning('httpcore.RemoteProtocolError: retry', exc_info=e)
243− # retry
244− except (httpcore.ProtocolError, httpcore.NetworkError) as e:
245− await close_connections_for_url(self._pool, url)
246− raise e
247−
248−
249−def get_transport_for_socks_proxy(verify, local_address, proxy_url):
250− global LOOP, LIMITS, TRANSPORT_KWARGS
251− # support socks5h (requests compatibility):
252− # https://requests.readthedocs.io/en/master/user/advanced/#socks
253− # socks5:// hostname is resolved on client side
254− # socks5h:// hostname is resolved on proxy side
255− rdns = False
256− socks5h = 'socks5h://'
257− if proxy_url.startswith(socks5h):
258− proxy_url = 'socks5://' + proxy_url[len(socks5h):]
259− rdns = True
260−
261− proxy_type, proxy_host, proxy_port, proxy_username, proxy_password = parse_proxy_url(proxy_url)
262−
263− return AsyncProxyTransportFixed(proxy_type=proxy_type, proxy_host=proxy_host, proxy_port=proxy_port,
264− username=proxy_username, password=proxy_password,
265− rdns=rdns,
266− loop=LOOP,
267− verify=verify,
268− local_address=local_address,
269− max_connections=LIMITS.max_connections,
270− max_keepalive_connections=LIMITS.max_keepalive_connections,
271− keepalive_expiry=LIMITS.keepalive_expiry,
272− **TRANSPORT_KWARGS)
273−
274−
275−def get_transport(verify, local_address, proxy_url):
276− global LIMITS
277− return AsyncHTTPTransportFixed(verify=verify,
278− local_address=local_address,
279− limits=LIMITS,
280− proxy=httpx._config.Proxy(proxy_url) if proxy_url else None,
281− **TRANSPORT_KWARGS)
282−
283−
284−def iter_proxies(proxies):
285− # https://www.python-httpx.org/compatibility/#proxy-keys
286− if isinstance(proxies, str):
287− yield 'all://', proxies
288− elif isinstance(proxies, dict):
289− for pattern, proxy_url in proxies.items():
290− pattern = PROXY_PATTERN_MAPPING.get(pattern, pattern)
291− yield pattern, proxy_url
292−
293−
294−def new_client(verify, local_address, proxies, max_redirects, enable_http):
295− # See https://www.python-httpx.org/advanced/#routing
296− mounts = {}
297− for pattern, proxy_url in iter_proxies(proxies):
298− if not enable_http and (pattern == 'http' or pattern.startswith('http://')):
299− continue
300− if proxy_url.startswith('socks4://') \
301− or proxy_url.startswith('socks5://') \
302− or proxy_url.startswith('socks5h://'):
303− mounts[pattern] = get_transport_for_socks_proxy(verify, local_address, proxy_url)
304− else:
305− mounts[pattern] = get_transport(verify, local_address, proxy_url)
306−
307− if not enable_http:
308− mounts['http://'] = AsyncHTTPTransportNoHttp()
309−
310− transport = get_transport(verify, local_address, None)
311− return httpx.AsyncClient(transport=transport, mounts=mounts, max_redirects=max_redirects)
312−
313−
314−def get_client(verify, local_address, proxies, max_redirects, allow_http):
315− global CLIENTS
316− key = (verify, local_address, repr(proxies), max_redirects, allow_http)
317− if key not in CLIENTS:
318− CLIENTS[key] = new_client(verify, local_address, proxies, max_redirects, allow_http)
319− return CLIENTS[key]
320−
321−
322−async def send_request(method, url, enable_http, kwargs):
323− if isinstance(url, bytes):
324− url = url.decode()
325−
326− verify = kwargs.pop('verify', True)
327− local_address = next(LOCAL_ADDRESS_CYCLE)
328− proxies = kwargs.pop('proxies', None) or get_global_proxies()
329− max_redirects = kwargs.pop('max_redirects', DEFAULT_REDIRECT_LIMIT)
330−
331− client = get_client(verify, local_address, proxies, max_redirects, enable_http)
332− response = await client.request(method.upper(), url, **kwargs)
333−
334− # requests compatibility
335− # see also https://www.python-httpx.org/compatibility/#checking-for-4xx5xx-responses
336− response.ok = not response.is_error
337−
338− return response
339−
340−
341−def request(method, url, **kwargs):
342− """same as requests/requests/api.py request(...)"""
343− time_before_request = time()
344−
345− # timeout
346− if 'timeout' in kwargs:
347− timeout = kwargs['timeout']
348− else:
349− timeout = getattr(THREADLOCAL, 'timeout', None)
350− if timeout is not None:
351− kwargs['timeout'] = timeout
352−
353− # raise_for_error
354− check_for_httperror = True
355− if 'raise_for_httperror' in kwargs:
356− check_for_httperror = kwargs['raise_for_httperror']
357− del kwargs['raise_for_httperror']
358−
359− # do request
360− future = asyncio.run_coroutine_threadsafe(send_request(method, url, get_enable_http_protocol(), kwargs), LOOP)
361− try:
362− if timeout:
363− timeout += 0.2 # overhead
364− start_time = getattr(THREADLOCAL, 'start_time', time_before_request)
365− if start_time:
366− timeout -= time() - start_time
367−
368− response = future.result(timeout or 120)
369− except concurrent.futures.TimeoutError as e:
370− raise httpx.TimeoutException('Timeout', request=None) from e
371−
372− # update total_time.
373− # See get_time_for_thread() and reset_time_for_thread()
374− if hasattr(THREADLOCAL, 'total_time'):
375− time_after_request = time()
376− THREADLOCAL.total_time += time_after_request - time_before_request
377−
378− # raise an exception
379− if check_for_httperror:
380− raise_for_httperror(response)
381−
382− return response
383−
384−
385−async def stream_chunk_to_queue(method, url, q, **kwargs):
386− verify = kwargs.pop('verify', True)
387− local_address = next(LOCAL_ADDRESS_CYCLE)
388− proxies = kwargs.pop('proxies', None) or get_global_proxies()
389− # "30" from requests:
390− # https://github.com/psf/requests/blob/8c211a96cdbe9fe320d63d9e1ae15c5c07e179f8/requests/models.py#L55
391− max_redirects = kwargs.pop('max_redirects', 30)
392− client = get_client(verify, local_address, proxies, max_redirects, True)
393− try:
394− async with client.stream(method, url, **kwargs) as response:
395− q.put(response)
396− async for chunk in response.aiter_bytes(65536):
397− if len(chunk) > 0:
398− q.put(chunk)
399− except (httpx.HTTPError, OSError, h2.exceptions.ProtocolError) as e:
400− q.put(e)
401− finally:
402− q.put(None)
403−
404−
405−def stream(method, url, **kwargs):
406− """Replace httpx.stream.
407−
408− Usage:
409− stream = poolrequests.stream(...)
410− response = next(stream)
411− for chunk in stream:
412− ...
413−
414− httpx.Client.stream requires to write the httpx.HTTPTransport version of the
415− the httpx.AsyncHTTPTransport declared above.
416− """
417− q = SimpleQueue()
418− future = asyncio.run_coroutine_threadsafe(stream_chunk_to_queue(method, url, q, **kwargs), LOOP)
419− chunk_or_exception = q.get(timeout=60)
420− while chunk_or_exception is not None:
421− if isinstance(chunk_or_exception, Exception):
422− raise chunk_or_exception
423− yield chunk_or_exception
424− chunk_or_exception = q.get(timeout=60)
425− return future.result()
426−
427−
428−def get(url, **kwargs):
429− kwargs.setdefault('allow_redirects', True)
430− return request('get', url, **kwargs)
431−
432−
433−def options(url, **kwargs):
434− kwargs.setdefault('allow_redirects', True)
435− return request('options', url, **kwargs)
436−
437−
438−def head(url, **kwargs):
439− kwargs.setdefault('allow_redirects', False)
440− return request('head', url, **kwargs)
441−
442−
443−def post(url, data=None, **kwargs):
444− return request('post', url, data=data, **kwargs)
445−
446−
447−def put(url, data=None, **kwargs):
448− return request('put', url, data=data, **kwargs)
449−
450−
451−def patch(url, data=None, **kwargs):
452− return request('patch', url, data=data, **kwargs)
453−
454−
455−def delete(url, **kwargs):
456− return request('delete', url, **kwargs)
457−
458−
459−def init():
460− # log
461− for logger_name in ('hpack.hpack', 'hpack.table'):
462− logging.getLogger(logger_name).setLevel(logging.WARNING)
463−
464− # loop
465− def loop_thread():
466− global LOOP
467− LOOP = asyncio.new_event_loop()
468− LOOP.run_forever()
469−
470− th = threading.Thread(
471− target=loop_thread,
472− name='asyncio_loop',
473− daemon=True,
474− )
475− th.start()
476−
477−
478−@atexit.register
479−def done():
480− """Close all HTTP client
481−
482− Avoid a warning at exit
483− see https://github.com/encode/httpx/blob/1a6e254f72d9fd5694a1c10a28927e193ab4f76b/httpx/_client.py#L1785
484− """
485− global LOOP
486−
487− async def close_client(client):
488− try:
489− await client.aclose()
490− except httpx.HTTPError:
491− pass
492−
493− async def close_clients():
494− await asyncio.gather(*[close_client(client) for client in CLIENTS.values()], return_exceptions=False)
495− future = asyncio.run_coroutine_threadsafe(close_clients(), LOOP)
496− # wait 3 seconds to close the HTTP clients
497− future.result(3)
498−
499−
500−init()
501−
502−
503−# ## TEMPORARY DEBUG ##
504−
505−
506−def debug_connection(connection):
507− now = LOOP.time()
508− expired = (connection.state == httpcore._async.base.ConnectionState.IDLE
509− and connection.expires_at is not None
510− and now >= connection.expires_at)
511− return connection.info()\
512− + (', connect_failed' if connection.connect_failed else '')\
513− + (', expired' if expired else '')
514−
515−
516−def debug_origin(origin):
517− return origin[0].decode() + '://' + origin[1].decode() + ':' + str(origin[2])
518−
519−
520−def debug_transport(transport):
521− result = {
522− '__class__': str(transport.__class__.__name__)
523− }
524− if isinstance(transport, (httpx.AsyncHTTPTransport, AsyncHTTPTransportFixed)):
525− pool = transport._pool
526− result['__pool_class__'] = str(pool.__class__.__name__)
527− if isinstance(pool, httpcore.AsyncConnectionPool):
528− for origin, connections in pool._connections.items():
529− result[debug_origin(origin)] = [debug_connection(connection) for connection in connections]
530− return result
531− elif isinstance(transport, AsyncProxyTransportFixed):
532− for origin, connections in transport._connections.items():
533− result[debug_origin(origin)] = [debug_connection(connection) for connection in connections]
534− return result
535− return result
536−
537−
538−def debug_asyncclient(client, key=None):
539− result = {}
540− if key:
541− result['__key__'] = [k if isinstance(k, (str, int, float, bool, type(None))) else repr(k) for k in key]
542− result['__default__'] = debug_transport(client._transport)
543− for urlpattern, transport in client._mounts.items():
544− result[urlpattern.pattern] = debug_transport(transport)
545− return result
546−
547−
548−def debug_asyncclients():
549− global CLIENTS
550− return [debug_asyncclient(client, key) for key, client in CLIENTS.items()]