3866429906

Build Type

push

github-actions

Committed by funilrys

Commit Message

fixup! Introduction of the support for postgresql DBs.

Run Details

11160 of 11683 relevant lines covered (95.52%)

14.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

20.65

/PyFunceble/query/requests/adapter/base.py

"""
The tool to check the availability or syntax of domain, IP or URL.

::


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides the base of all our adapter.

Author:
    Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom

Special thanks:
    https://pyfunceble.github.io/#/special-thanks

Contributors:
    https://pyfunceble.github.io/#/contributors

Project link:
    https://github.com/funilrys/PyFunceble

Project documentation:
    https://pyfunceble.readthedocs.io/en/dev/

Project homepage:
    https://pyfunceble.github.io/

License:
::


    Copyright 2017, 2018, 2019, 2020, 2022 Nissar Chababy

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

from typing import Optional

import requests.adapters
import requests.models

import PyFunceble.storage
from PyFunceble.checker.syntax.ip import IPSyntaxChecker
from PyFunceble.query.dns.query_tool import DNSQueryTool


class RequestAdapterBase(requests.adapters.HTTPAdapter):
    """
    Extends the built-in HTTP adapater and acts as a base for all our own
    adapter.
    """

    resolving_cache: dict = {}
    resolving_use_cache: bool = False
    timeout: float = 5.0
    proxy_pattern: dict = {}

    def __init__(self, *args, **kwargs):
        if "timeout" in kwargs:
            self.timeout = float(kwargs["timeout"])
            del kwargs["timeout"]

        if "max_retries" in kwargs:
            kwargs["max_retries"] = requests.adapters.Retry(
                total=kwargs["max_retries"], respect_retry_after_header=False
            )

        if "dns_query_tool" in kwargs:
            self.dns_query_tool = kwargs["dns_query_tool"]
            del kwargs["dns_query_tool"]
        else:
            self.dns_query_tool = DNSQueryTool()

        if "proxy_pattern" in kwargs:
            self.proxy_pattern = kwargs["proxy_pattern"]
            del kwargs["proxy_pattern"]
        else:
            self.proxy_pattern = {}

        super().__init__(*args, **kwargs)

    @staticmethod
    def fake_response() -> requests.models.Response:
        """
        Provides the fake response that is provided when we couldn't resolve the
        given domain.
        """

        raise PyFunceble.factory.Requester.exceptions.ConnectionError(
            "Could not resolve."
        )

    @staticmethod
    def extract_extension(subject: str) -> Optional[str]:
        """
        Provides the extension of the given subject.

        .. versionchanged:: 4.1.1.dev
            Handle the case that the given subject does not have a `.` (point).

        :param str subject:
            The subject to get extract the extension from.

        :raise TypeError:
            When the given :code:`subject` is not a :py:class:`str`.
        :raise ValueError:
            When the given :code:`subject` is an empty :py:class:`str`.
        """

        if not subject or "." not in subject:
            return None

        if subject.endswith("."):
            # Absolute needs a little correction.
            last_point = subject[:-1].rfind(".")
        else:
            last_point = subject.rindex(".")

        extension = subject[last_point + 1 :]

        if extension.endswith("."):
            return extension[:-1]
        return extension

    def fetch_proxy_from_pattern(self, subject: str) -> dict:
        """
        Provides the proxy settings to use for the given subject.

        .. versionchanged:: 4.1.1.dev
            Handle the case that the given subject has no extension/TLD.

        :param str subject:
            The subject to work with.

        :raise TypeError:
            When the given :code:`subject` is not a :py:class:`str`.
        :raise ValueError:
            When the given :code:`subject` is an empty :py:class:`str`.
        """

        def correct_input(pattern_input: dict) -> dict:
            result = {}

            if "http" in pattern_input and pattern_input["http"]:
                result["http"] = pattern_input["http"]

            if "https" in pattern_input and pattern_input["https"]:
                result["https"] = pattern_input["https"]

            if "http" in result and "https" not in result:
                result["https"] = result["http"]

            if "https" in result and "http" not in result:
                result["http"] = result["https"]

            return result

        extension = self.extract_extension(subject)

        proxies = {}

        if extension and "rules" in self.proxy_pattern:
            for rule in self.proxy_pattern["rules"]:
                local_proxy = {}

                if "http" in rule and rule["http"]:
                    local_proxy["http"] = rule["http"]
                if "https" in rule and rule["https"]:
                    local_proxy["https"] = rule["https"]

                if not local_proxy:
                    continue

                if "tld" in rule and extension in rule["tld"]:
                    proxies = correct_input(local_proxy)
                    break

        if not proxies and "global" in self.proxy_pattern:
            proxies = correct_input(self.proxy_pattern["global"])

        return proxies

    def resolve_with_cache(self, hostname: str) -> Optional[str]:
        """
        Try to resolve using an internal cache.
        """

        if hostname not in self.resolving_cache:
            self.resolving_cache[hostname] = self.resolve_without_cache(hostname)

        return self.resolving_cache[hostname]

    def resolve_without_cache(self, hostname: str) -> Optional[str]:
        """
        Resolves the IP of the given hostname.

        :param hostname:
            The hostname to get resolve.
        """

        def get_last_cname(subject: str, recursion_depth: int = 60) -> Optional[str]:
            """
            Given a subject, this function tries to query the CNAME until there
            is none.

            :param subject:
                The first subject.
            """

            last_cname_result = []
            last_cname_new_subject = subject

            depth = 0

            while depth < recursion_depth:
                local_last_cname_result = (
                    self.dns_query_tool.set_query_record_type("CNAME")
                    .set_subject(last_cname_new_subject)
                    .query()
                )

                depth += 1

                if any(x in last_cname_result for x in local_last_cname_result):
                    break

                last_cname_result.extend(local_last_cname_result)

                if local_last_cname_result:
                    last_cname_new_subject = local_last_cname_result[0]
                else:
                    break

            try:
                return last_cname_result[-1]
            except IndexError:
                return None

        result = set()

        if not IPSyntaxChecker(hostname).is_valid():
            last_cname = get_last_cname(hostname)

            if last_cname:
                result.update(
                    self.dns_query_tool.set_query_record_type("A")
                    .set_subject(last_cname)
                    .query()
                )
            else:
                result.update(
                    self.dns_query_tool.set_query_record_type("A")
                    .set_subject(hostname)
                    .query()
                )
        else:
            result.add(hostname)

        if result:
            return result.pop()
        return None

    def resolve(self, hostname: str) -> Optional[str]:
        """
        Resolves with the prefered method.
        """

        if hostname:
            if self.resolving_use_cache:
                return self.resolve_with_cache(hostname)
            return self.resolve_without_cache(hostname)
        return None

1	"""
2	The tool to check the availability or syntax of domain, IP or URL.
3
4	::
5
6
7	██████╗ ██╗ ██╗███████╗██╗ ██╗███╗ ██╗ ██████╗███████╗██████╗ ██╗ ███████╗
8	██╔══██╗╚██╗ ██╔╝██╔════╝██║ ██║████╗ ██║██╔════╝██╔════╝██╔══██╗██║ ██╔════╝
9	██████╔╝ ╚████╔╝ █████╗ ██║ ██║██╔██╗ ██║██║ █████╗ ██████╔╝██║ █████╗
10	██╔═══╝ ╚██╔╝ ██╔══╝ ██║ ██║██║╚██╗██║██║ ██╔══╝ ██╔══██╗██║ ██╔══╝
11	██║ ██║ ██║ ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
12	╚═╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝
13
14	Provides the base of all our adapter.
15
16	Author:
17	Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom
18
19	Special thanks:
20	https://pyfunceble.github.io/#/special-thanks
21
22	Contributors:
23	https://pyfunceble.github.io/#/contributors
24
25	Project link:
26	https://github.com/funilrys/PyFunceble
27
28	Project documentation:
29	https://pyfunceble.readthedocs.io/en/dev/
30
31	Project homepage:
32	https://pyfunceble.github.io/
33
34	License:
35	::
36
37
38	Copyright 2017, 2018, 2019, 2020, 2022 Nissar Chababy
39
40	Licensed under the Apache License, Version 2.0 (the "License");
41	you may not use this file except in compliance with the License.
42	You may obtain a copy of the License at
43
44	http://www.apache.org/licenses/LICENSE-2.0
45
46	Unless required by applicable law or agreed to in writing, software
47	distributed under the License is distributed on an "AS IS" BASIS,
48	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
49	See the License for the specific language governing permissions and
50	limitations under the License.
51	"""
52
53	from typing import Optional	15✔
54
55	import requests.adapters	15✔
56	import requests.models	15✔
57
58	import PyFunceble.storage	15✔
59	from PyFunceble.checker.syntax.ip import IPSyntaxChecker	15✔
60	from PyFunceble.query.dns.query_tool import DNSQueryTool	15✔
61
62
63	class RequestAdapterBase(requests.adapters.HTTPAdapter):	15✔
64	"""
65	Extends the built-in HTTP adapater and acts as a base for all our own
66	adapter.
67	"""
68
69	resolving_cache: dict = {}	15✔
70	resolving_use_cache: bool = False	15✔
71	timeout: float = 5.0	15✔
72	proxy_pattern: dict = {}	15✔
73
74	def __init__(self, args, *kwargs):
75	if "timeout" in kwargs:
76	self.timeout = float(kwargs["timeout"])
77	del kwargs["timeout"]
78
79	if "max_retries" in kwargs:
80	kwargs["max_retries"] = requests.adapters.Retry(
81	total=kwargs["max_retries"], respect_retry_after_header=False
82	)
83
84	if "dns_query_tool" in kwargs:
85	self.dns_query_tool = kwargs["dns_query_tool"]
86	del kwargs["dns_query_tool"]
87	else:
88	self.dns_query_tool = DNSQueryTool()
89
90	if "proxy_pattern" in kwargs:
91	self.proxy_pattern = kwargs["proxy_pattern"]
92	del kwargs["proxy_pattern"]
93	else:
94	self.proxy_pattern = {}
95
96	super().__init__(args, *kwargs)
97
98	@staticmethod	15✔
99	def fake_response() -> requests.models.Response:	15✔
100	"""
101	Provides the fake response that is provided when we couldn't resolve the
102	given domain.
103	"""
104
105	raise PyFunceble.factory.Requester.exceptions.ConnectionError(	×
106	"Could not resolve."
107	)
108
109	@staticmethod	15✔
110	def extract_extension(subject: str) -> Optional[str]:	15✔
111	"""
112	Provides the extension of the given subject.
113
114	.. versionchanged:: 4.1.1.dev
115	Handle the case that the given subject does not have a `.` (point).
116
117	:param str subject:
118	The subject to get extract the extension from.
119
120	:raise TypeError:
121	When the given :code:`subject` is not a :py:class:`str`.
122	:raise ValueError:
123	When the given :code:`subject` is an empty :py:class:`str`.
124	"""
125
126	if not subject or "." not in subject:	×
127	return None	×
128
129	if subject.endswith("."):	×
130	# Absolute needs a little correction.
131	last_point = subject[:-1].rfind(".")	×
132	else:
133	last_point = subject.rindex(".")	×
134
135	extension = subject[last_point + 1 :]	×
136
137	if extension.endswith("."):	×
138	return extension[:-1]	×
139	return extension	×
140
141	def fetch_proxy_from_pattern(self, subject: str) -> dict:	15✔
142	"""
143	Provides the proxy settings to use for the given subject.
144
145	.. versionchanged:: 4.1.1.dev
146	Handle the case that the given subject has no extension/TLD.
147
148	:param str subject:
149	The subject to work with.
150
151	:raise TypeError:
152	When the given :code:`subject` is not a :py:class:`str`.
153	:raise ValueError:
154	When the given :code:`subject` is an empty :py:class:`str`.
155	"""
156
157	def correct_input(pattern_input: dict) -> dict:	×
158	result = {}	×
159
160	if "http" in pattern_input and pattern_input["http"]:	×
161	result["http"] = pattern_input["http"]	×
162
163	if "https" in pattern_input and pattern_input["https"]:	×
164	result["https"] = pattern_input["https"]	×
165
166	if "http" in result and "https" not in result:	×
167	result["https"] = result["http"]	×
168
169	if "https" in result and "http" not in result:	×
170	result["http"] = result["https"]	×
171
172	return result	×
173
174	extension = self.extract_extension(subject)	×
175
176	proxies = {}	×
177
178	if extension and "rules" in self.proxy_pattern:	×
179	for rule in self.proxy_pattern["rules"]:	×
180	local_proxy = {}	×
181
182	if "http" in rule and rule["http"]:	×
183	local_proxy["http"] = rule["http"]	×
184	if "https" in rule and rule["https"]:	×
185	local_proxy["https"] = rule["https"]	×
186
187	if not local_proxy:	×
188	continue	×
189
190	if "tld" in rule and extension in rule["tld"]:	×
191	proxies = correct_input(local_proxy)	×
192	break	×
193
194	if not proxies and "global" in self.proxy_pattern:	×
195	proxies = correct_input(self.proxy_pattern["global"])	×
196
197	return proxies	×
198
199	def resolve_with_cache(self, hostname: str) -> Optional[str]:	15✔
200	"""
201	Try to resolve using an internal cache.
202	"""
203
204	if hostname not in self.resolving_cache:	×
205	self.resolving_cache[hostname] = self.resolve_without_cache(hostname)	×
206
207	return self.resolving_cache[hostname]	×
208
209	def resolve_without_cache(self, hostname: str) -> Optional[str]:	15✔
210	"""
211	Resolves the IP of the given hostname.
212
213	:param hostname:
214	The hostname to get resolve.
215	"""
216
217	def get_last_cname(subject: str, recursion_depth: int = 60) -> Optional[str]:	×
218	"""
219	Given a subject, this function tries to query the CNAME until there
220	is none.
221
222	:param subject:
223	The first subject.
224	"""
225
226	last_cname_result = []	×
227	last_cname_new_subject = subject	×
228
229	depth = 0	×
230
231	while depth < recursion_depth:	×
232	local_last_cname_result = (	×
233	self.dns_query_tool.set_query_record_type("CNAME")
234	.set_subject(last_cname_new_subject)
235	.query()
236	)
237
238	depth += 1	×
239
240	if any(x in last_cname_result for x in local_last_cname_result):	×
241	break	×
242
243	last_cname_result.extend(local_last_cname_result)	×
244
245	if local_last_cname_result:	×
246	last_cname_new_subject = local_last_cname_result[0]	×
247	else:
248	break	×
249
250	try:	×
251	return last_cname_result[-1]	×
252	except IndexError:	×
253	return None	×
254
255	result = set()	×
256
257	if not IPSyntaxChecker(hostname).is_valid():	×
258	last_cname = get_last_cname(hostname)	×
259
260	if last_cname:	×
261	result.update(	×
262	self.dns_query_tool.set_query_record_type("A")
263	.set_subject(last_cname)
264	.query()
265	)
266	else:
267	result.update(	×
268	self.dns_query_tool.set_query_record_type("A")
269	.set_subject(hostname)
270	.query()
271	)
272	else:
273	result.add(hostname)	×
274
275	if result:	×
276	return result.pop()	×
277	return None	×
278
279	def resolve(self, hostname: str) -> Optional[str]:	15✔
280	"""
281	Resolves with the prefered method.
282	"""
283
284	if hostname:	×
285	if self.resolving_use_cache:	×
286	return self.resolve_with_cache(hostname)	×
287	return self.resolve_without_cache(hostname)	×
288	return None	×

funilrys / PyFunceble / 3866429906

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous