@@ -0,0 +1,303 @@
1+# SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+import atexit
4+import asyncio
5+import ipaddress
6+from itertools import cycle
7+
8+import httpx
9+
10+from .client import new_client, get_loop
11+
12+
13+DEFAULT_NAME = '__DEFAULT__'
14+NETWORKS = {}
15+# requests compatibility when reading proxy settings from settings.yml
16+PROXY_PATTERN_MAPPING = {
17+ 'http': 'http://',
18+ 'https': 'https://',
19+ 'socks4': 'socks4://',
20+ 'socks5': 'socks5://',
21+ 'socks5h': 'socks5h://',
22+ 'http:': 'http://',
23+ 'https:': 'https://',
24+ 'socks4:': 'socks4://',
25+ 'socks5:': 'socks5://',
26+ 'socks5h:': 'socks5h://',
27+}
28+
29+ADDRESS_MAPPING = {
30+ 'ipv4': '0.0.0.0',
31+ 'ipv6': '::'
32+}
33+
34+
35+class Network:
36+
37+ __slots__ = ('enable_http', 'verify', 'enable_http2',
38+ 'max_connections', 'max_keepalive_connections', 'keepalive_expiry',
39+ 'local_addresses', 'proxies', 'max_redirects', 'retries', 'retry_on_http_error',
40+ '_local_addresses_cycle', '_proxies_cycle', '_clients')
41+
42+ def __init__(self,
43+ enable_http=True,
44+ verify=True,
45+ enable_http2=False,
46+ max_connections=None,
47+ max_keepalive_connections=None,
48+ keepalive_expiry=None,
49+ proxies=None,
50+ local_addresses=None,
51+ retries=0,
52+ retry_on_http_error=None,
53+ max_redirects=30):
54+ self.enable_http = enable_http
55+ self.verify = verify
56+ self.enable_http2 = enable_http2
57+ self.max_connections = max_connections
58+ self.max_keepalive_connections = max_keepalive_connections
59+ self.keepalive_expiry = keepalive_expiry
60+ self.proxies = proxies
61+ self.local_addresses = local_addresses
62+ self.retries = retries
63+ self.retry_on_http_error = retry_on_http_error
64+ self.max_redirects = max_redirects
65+ self._local_addresses_cycle = self.get_ipaddress_cycle()
66+ self._proxies_cycle = self.get_proxy_cycles()
67+ self._clients = {}
68+ self.check_parameters()
69+
70+ def check_parameters(self):
71+ for address in self.iter_ipaddresses():
72+ if '/' in address:
73+ ipaddress.ip_network(address, False)
74+ else:
75+ ipaddress.ip_address(address)
76+
77+ if self.proxies is not None and not isinstance(self.proxies, (str, dict)):
78+ raise ValueError('proxies type has to be str, dict or None')
79+
80+ def iter_ipaddresses(self):
81+ local_addresses = self.local_addresses
82+ if not local_addresses:
83+ return
84+ elif isinstance(local_addresses, str):
85+ local_addresses = [local_addresses]
86+ for address in local_addresses:
87+ yield address
88+
89+ def get_ipaddress_cycle(self):
90+ while True:
91+ count = 0
92+ for address in self.iter_ipaddresses():
93+ if '/' in address:
94+ for a in ipaddress.ip_network(address, False).hosts():
95+ yield str(a)
96+ count += 1
97+ else:
98+ a = ipaddress.ip_address(address)
99+ yield str(a)
100+ count += 1
101+ if count == 0:
102+ yield None
103+
104+ def iter_proxies(self):
105+ if not self.proxies:
106+ return
107+ # https://www.python-httpx.org/compatibility/#proxy-keys
108+ if isinstance(self.proxies, str):
109+ yield 'all://', [self.proxies]
110+ else:
111+ for pattern, proxy_url in self.proxies.items():
112+ pattern = PROXY_PATTERN_MAPPING.get(pattern, pattern)
113+ if isinstance(proxy_url, str):
114+ proxy_url = [proxy_url]
115+ yield pattern, proxy_url
116+
117+ def get_proxy_cycles(self):
118+ proxy_settings = {}
119+ for pattern, proxy_urls in self.iter_proxies():
120+ proxy_settings[pattern] = cycle(proxy_urls)
121+ while True:
122+ yield tuple((pattern, next(proxy_url_cycle)) for pattern, proxy_url_cycle in proxy_settings.items())
123+
124+ def get_client(self, verify=None, max_redirects=None):
125+ verify = self.verify if verify is None else verify
126+ max_redirects = self.max_redirects if max_redirects is None else max_redirects
127+ local_address = next(self._local_addresses_cycle)
128+ proxies = next(self._proxies_cycle) # is a tuple so it can be part of the key
129+ key = (verify, max_redirects, local_address, proxies)
130+ if key not in self._clients or self._clients[key].is_closed:
131+ self._clients[key] = new_client(self.enable_http,
132+ verify,
133+ self.enable_http2,
134+ self.max_connections,
135+ self.max_keepalive_connections,
136+ self.keepalive_expiry,
137+ dict(proxies),
138+ local_address,
139+ 0,
140+ max_redirects)
141+ return self._clients[key]
142+
143+ async def aclose(self):
144+ async def close_client(client):
145+ try:
146+ await client.aclose()
147+ except httpx.HTTPError:
148+ pass
149+ await asyncio.gather(*[close_client(client) for client in self._clients.values()], return_exceptions=False)
150+
151+ @staticmethod
152+ def get_kwargs_clients(kwargs):
153+ kwargs_clients = {}
154+ if 'verify' in kwargs:
155+ kwargs_clients['verify'] = kwargs.pop('verify')
156+ if 'max_redirects' in kwargs:
157+ kwargs_clients['max_redirects'] = kwargs.pop('max_redirects')
158+ return kwargs_clients
159+
160+ def is_valid_respones(self, response):
161+ if (self.retry_on_http_error is True and 400 <= response.status_code <= 599) \
162+ or (isinstance(self.retry_on_http_error, list) and response.status_code in self.retry_on_http_error) \
163+ or (isinstance(self.retry_on_http_error, int) and response.status_code == self.retry_on_http_error):
164+ return False
165+ return True
166+
167+ async def request(self, method, url, **kwargs):
168+ retries = self.retries
169+ while retries >= 0: # pragma: no cover
170+ kwargs_clients = Network.get_kwargs_clients(kwargs)
171+ client = self.get_client(**kwargs_clients)
172+ try:
173+ response = await client.request(method, url, **kwargs)
174+ if self.is_valid_respones(response) or retries <= 0:
175+ return response
176+ except (httpx.RequestError, httpx.HTTPStatusError) as e:
177+ if retries <= 0:
178+ raise e
179+ retries -= 1
180+
181+ def stream(self, method, url, **kwargs):
182+ retries = self.retries
183+ while retries >= 0: # pragma: no cover
184+ kwargs_clients = Network.get_kwargs_clients(kwargs)
185+ client = self.get_client(**kwargs_clients)
186+ try:
187+ response = client.stream(method, url, **kwargs)
188+ if self.is_valid_respones(response) or retries <= 0:
189+ return response
190+ except (httpx.RequestError, httpx.HTTPStatusError) as e:
191+ if retries <= 0:
192+ raise e
193+ retries -= 1
194+
195+ @classmethod
196+ async def aclose_all(cls):
197+ await asyncio.gather(*[network.aclose() for network in NETWORKS.values()], return_exceptions=False)
198+
199+
200+def get_network(name=None):
201+ global NETWORKS
202+ return NETWORKS[name or DEFAULT_NAME]
203+
204+
205+def initialize(settings_engines=None, settings_outgoing=None):
206+ from searx.engines import engines
207+ from searx import settings
208+
209+ global NETWORKS
210+
211+ settings_engines = settings_engines or settings.get('engines')
212+ settings_outgoing = settings_outgoing or settings.get('outgoing')
213+
214+ # default parameters for AsyncHTTPTransport
215+ # see https://github.com/encode/httpx/blob/e05a5372eb6172287458b37447c30f650047e1b8/httpx/_transports/default.py#L108-L121 # noqa
216+ default_params = {
217+ 'enable_http': False,
218+ 'verify': True,
219+ 'enable_http2': settings_outgoing.get('enable_http2', True),
220+ # Magic number kept from previous code
221+ 'max_connections': settings_outgoing.get('pool_connections', 100),
222+ # Picked from constructor
223+ 'max_keepalive_connections': settings_outgoing.get('pool_maxsize', 10),
224+ #
225+ 'keepalive_expiry': settings_outgoing.get('keepalive_expiry', 5.0),
226+ 'local_addresses': settings_outgoing.get('source_ips'),
227+ 'proxies': settings_outgoing.get('proxies'),
228+ # default maximum redirect
229+ # from https://github.com/psf/requests/blob/8c211a96cdbe9fe320d63d9e1ae15c5c07e179f8/requests/models.py#L55
230+ 'max_redirects': settings_outgoing.get('max_redirects', 30),
231+ #
232+ 'retries': settings_outgoing.get('retries', 0),
233+ 'retry_on_http_error': None,
234+ }
235+
236+ def new_network(params):
237+ nonlocal default_params
238+ result = {}
239+ result.update(default_params)
240+ result.update(params)
241+ return Network(**result)
242+
243+ def iter_networks():
244+ nonlocal settings_engines
245+ for engine_spec in settings_engines:
246+ engine_name = engine_spec['name']
247+ engine = engines.get(engine_name)
248+ if engine is None:
249+ continue
250+ network = getattr(engine, 'network', None)
251+ yield engine_name, engine, network
252+
253+ if NETWORKS:
254+ done()
255+ NETWORKS.clear()
256+ NETWORKS[DEFAULT_NAME] = new_network({})
257+ NETWORKS['ipv4'] = new_network({'local_addresses': '0.0.0.0'})
258+ NETWORKS['ipv6'] = new_network({'local_addresses': '::'})
259+
260+ # define networks from outgoing.networks
261+ for network_name, network in settings_outgoing.get('networks', {}).items():
262+ NETWORKS[network_name] = new_network(network)
263+
264+ # define networks from engines.[i].network (except references)
265+ for engine_name, engine, network in iter_networks():
266+ if network is None:
267+ network = {}
268+ for attribute_name, attribute_value in default_params.items():
269+ if hasattr(engine, attribute_name):
270+ network[attribute_name] = getattr(engine, attribute_name)
271+ else:
272+ network[attribute_name] = attribute_value
273+ NETWORKS[engine_name] = new_network(network)
274+ elif isinstance(network, dict):
275+ NETWORKS[engine_name] = new_network(network)
276+
277+ # define networks from engines.[i].network (references)
278+ for engine_name, engine, network in iter_networks():
279+ if isinstance(network, str):
280+ NETWORKS[engine_name] = NETWORKS[network]
281+
282+
283+@atexit.register
284+def done():
285+ """Close all HTTP client
286+
287+ Avoid a warning at exit
288+ see https://github.com/encode/httpx/blob/1a6e254f72d9fd5694a1c10a28927e193ab4f76b/httpx/_client.py#L1785
289+
290+ Note: since Network.aclose has to be async, it is not possible to call this method on Network.__del__
291+ So Network.aclose is called here using atexit.register
292+ """
293+ try:
294+ loop = get_loop()
295+ if loop:
296+ future = asyncio.run_coroutine_threadsafe(Network.aclose_all(), loop)
297+ # wait 3 seconds to close the HTTP clients
298+ future.result(3)
299+ finally:
300+ NETWORKS.clear()
301+
302+
303+NETWORKS[DEFAULT_NAME] = Network()