Add switch for auth responses

asyncwhois/__init__.py

@@ -42,26 +42,29 @@
     "GeneralError",
     "QueryError",
 ]
-__version__ = "1.1.7"
+__version__ = "1.1.8"
 
 
 def whois(
     search_term: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address],
-    authoritative_only: bool = False,
+    authoritative_only: bool = False,  # todo: deprecate and remove this argument
+    find_authoritative_server: bool = True,
     ignore_not_found: bool = False,
     proxy_url: str = None,
     timeout: int = 10,
     tldextract_obj: TLDExtract = None,
-    max_depth: int = None,
 ) -> tuple[str, dict]:
     """
     Performs a WHOIS query for the given `search_term`. If `search_term` is or can be cast to an
     instance of `ipaddress.IPv4Address` or `ipaddress.IPv6Address` then an IP search is performed
     otherwise a DNS search is performed.
 
     :param search_term: Any domain, URL, IPv4, or IPv6
-    :param authoritative_only: If False (default), asyncwhois returns the entire WHOIS query chain,
+    :param authoritative_only: DEPRECATED - If False (default), asyncwhois returns the entire WHOIS query chain,
         otherwise if True, only the authoritative response is returned.
+    :param find_authoritative_server: This parameter only applies to domain queries. If True (default), asyncwhois
+        will attempt to find the authoritative response, otherwise if False, asyncwhois will only query the whois server
+        associated with the given TLD as specified in the IANA root db (`asyncwhois/servers/domains.py`).
     :param ignore_not_found:  If False (default), the `NotFoundError` exception is raised if the query output
         contains "no such domain" language. If True, asyncwhois will not raise `NotFoundError` exceptions.
     :param proxy_url: Optional SOCKS4 or SOCKS5 proxy url (e.g. 'socks5://host:port')
@@ -86,11 +89,11 @@ def whois(
         except (ipaddress.AddressValueError, ValueError):
             return DomainClient(
                 authoritative_only=authoritative_only,
+                find_authoritative_server=find_authoritative_server,
                 ignore_not_found=ignore_not_found,
                 proxy_url=proxy_url,
                 timeout=timeout,
                 tldextract_obj=tldextract_obj,
-                max_depth=max_depth,
             ).whois(search_term)
     else:
         return "", {}
@@ -142,21 +145,24 @@ def rdap(
 
 async def aio_whois(
     search_term: str,
-    authoritative_only: bool = False,
+    authoritative_only: bool = False,  # todo: deprecate and remove this argument
+    find_authoritative_server: bool = True,
     ignore_not_found: bool = False,
     proxy_url: str = None,
     timeout: int = 10,
     tldextract_obj: TLDExtract = None,
-    max_depth: int = None,
 ) -> tuple[str, dict]:
     """
     Performs a WHOIS query for the given `search_term`. If `search_term` is or can be cast to an
     instance of `ipaddress.IPv4Address` or `ipaddress.IPv6Address` then an IP search is performed
     otherwise a DNS search is performed.
 
     :param search_term: Any domain, URL, IPv4, or IPv6
-    :param authoritative_only: If False (default), asyncwhois returns the entire WHOIS query chain,
+    :param authoritative_only: DEPRECATED - If False (default), asyncwhois returns the entire WHOIS query chain,
         otherwise if True, only the authoritative response is returned.
+    :param find_authoritative_server: This parameter only applies to domain queries. If True (default), asyncwhois
+        will attempt to find the authoritative response, otherwise if False, asyncwhois will only query the whois server
+        associated with the given TLD as specified in the IANA root db (`asyncwhois/servers/domains.py`).
     :param ignore_not_found:  If False (default), the `NotFoundError` exception is raised if the query output
         contains "no such domain" language. If True, asyncwhois will not raise `NotFoundError` exceptions.
     :param proxy_url: Optional SOCKS4 or SOCKS5 proxy url (e.g. 'socks5://host:port')
@@ -185,7 +191,7 @@ async def aio_whois(
                 proxy_url=proxy_url,
                 timeout=timeout,
                 tldextract_obj=tldextract_obj,
-                max_depth=max_depth,
+                find_authoritative_server=find_authoritative_server,
             ).aio_whois(search_term)
     else:
         return "", {}

asyncwhois/client.py

@@ -49,20 +49,24 @@ class DomainClient(Client):
     def __init__(
         self,
         authoritative_only: bool = False,
+        find_authoritative_server: bool = True,
         ignore_not_found: bool = False,
         proxy_url: str = None,
         whodap_client: whodap.DNSClient = None,
         timeout: int = 10,
         tldextract_obj: TLDExtract = None,
-        max_depth: int = None,
     ):
         super().__init__(whodap_client)
         self.authoritative_only = authoritative_only
         self.ignore_not_found = ignore_not_found
         self.proxy_url = proxy_url
         self.timeout = timeout
         self.tldextract_obj = tldextract_obj
-        self.query_obj = DomainQuery(proxy_url=proxy_url, timeout=timeout, max_depth=max_depth)
+        self.query_obj = DomainQuery(
+            proxy_url=proxy_url,
+            timeout=timeout,
+            find_authoritative_server=find_authoritative_server,
+        )
         self.parse_obj = DomainParser(ignore_not_found=ignore_not_found)
 
     def _get_domain_components(self, domain: str) -> tuple[str, str, str]:

asyncwhois/query.py

@@ -19,11 +19,16 @@ class Query:
     refer_regex = r"refer: *(.+)"
     whois_server_regex = r".+ whois server: *(.+)"
 
-    def __init__(self, proxy_url: str = None, timeout: int = 10, max_depth: int = None):
+    def __init__(
+        self,
+        proxy_url: str = None,
+        timeout: int = 10,
+        find_authoritative_server: bool = True,
+    ):
         self.proxy_url = proxy_url
         self.timeout = timeout
-        self.max_depth = max_depth
-        
+        self.find_authoritative_server = find_authoritative_server
+
     @staticmethod
     def _find_match(regex: str, blob: str) -> str:
         match = ""
@@ -118,6 +123,16 @@ def run(self, search_term: str, server: str = None) -> list[str]:
             server_regex = self.whois_server_regex
         return self._do_query(server, data, server_regex, [])
 
+    @staticmethod
+    def _continue_querying(current_server: str, next_server: str) -> bool:
+        next_server = next_server.lower()
+        return (
+            next_server
+            and next_server != current_server
+            and not next_server.startswith("http")
+            and not next_server.startswith("www.")
+        )
+
     async def aio_run(self, search_term: str, server: str = None) -> list[str]:
         data = search_term + "\r\n"
         if not server:
@@ -131,7 +146,7 @@ async def aio_run(self, search_term: str, server: str = None) -> list[str]:
         return await self._aio_do_query(server, data, server_regex, [])
 
     def _do_query(
-        self, server: str, data: str, regex: str, chain: list[str], depth: int = 0
+        self, server: str, data: str, regex: str, chain: list[str]
     ) -> list[str]:
         """
         Recursively submits WHOIS queries until it reaches the Authoritative Server.
@@ -142,22 +157,16 @@ def _do_query(
             query_output = self._send_and_recv(conn, data)
             # save query chain
             chain.append(query_output)
-            # if max depth is reached, return the chain
-            if self.max_depth and depth >= self.max_depth:
-                return chain            
-            # parse response for the referred WHOIS server name
-            whois_server = self._find_match(regex, query_output)
-            whois_server = whois_server.lower()
-            if (
-                whois_server
-                and whois_server != server
-                and not whois_server.startswith("http")
-                and not whois_server.startswith("www.")
-            ):
-                # recursive call to find more authoritative server
-                chain = self._do_query(
-                    whois_server, data, self.whois_server_regex, chain, depth + 1
-                )
+            # if we should find the authoritative response,
+            # then parse the response for the next server
+            if self.find_authoritative_server:
+                # parse response for the referred WHOIS server name
+                whois_server = self._find_match(regex, query_output)
+                if self._continue_querying(server, whois_server):
+                    # recursive call to find more authoritative server
+                    chain = self._do_query(
+                        whois_server, data, self.whois_server_regex, chain
+                    )
         # return the WHOIS query chain
         return chain
 
@@ -175,23 +184,16 @@ async def _aio_do_query(
                 self._aio_send_and_recv(reader, writer, data), self.timeout
             )
             chain.append(query_output)
-            # if max depth is reached, return the chain
-            if self.max_depth is not None and depth >= self.max_depth:
-                return chain
-            # parse response for the referred WHOIS server name
-            whois_server = self._find_match(regex, query_output)
-            whois_server = whois_server.lower()
-            # check for another legitimate server name
-            if (
-                whois_server
-                and whois_server != server
-                and not whois_server.startswith("http")
-                and not whois_server.startswith("www.")
-            ):
-                # recursive call to find the authoritative server
-                chain = await self._aio_do_query(
-                    whois_server, data, self.whois_server_regex, chain, depth + 1
-                )
+            # if we should find the authoritative response,
+            # then parse the response for the next server
+            if self.find_authoritative_server:
+                # parse response for the referred WHOIS server name
+                whois_server = self._find_match(regex, query_output)
+                if self._continue_querying(server, whois_server):
+                    # recursive call to find more authoritative server
+                    chain = self._do_query(
+                        whois_server, data, self.whois_server_regex, chain
+                    )
         # return the WHOIS query chain
         return chain
 
@@ -202,9 +204,9 @@ def __init__(
         server: str = None,
         proxy_url: str = None,
         timeout: int = 10,
-        max_depth: int = None,
+        find_authoritative_server: bool = True,
     ):
-        super().__init__(proxy_url, timeout, max_depth)
+        super().__init__(proxy_url, timeout, find_authoritative_server)
         self.server = server
 
     @staticmethod