IDEV-1996: Implement domaindiscovery python wrapper endpoint.

Author: jbabac

domaintools/api.py

@@ -3,6 +3,7 @@
 from hmac import new as hmac
 import re
 
+from domaintools.constants import Endpoint, ENDPOINT_TO_SOURCE_MAP, OutputFormat
 from domaintools._version import current as version
 from domaintools.results import (
     GroupedIterable,
@@ -18,6 +19,8 @@
     filter_by_field,
     DTResultFilter,
 )
+from domaintools.utils import validate_feeds_required_parameters
+
 
 AVAILABLE_KEY_SIGN_HASHES = ["sha1", "sha256", "md5"]
 
@@ -1088,15 +1091,28 @@ def nad(self, **kwargs):
 
     def domainrdap(self, **kwargs):
         """Returns changes to global domain registration information, populated by the Registration Data Access Protocol (RDAP)"""
-        sessionID = kwargs.get("sessionID")
-        after = kwargs.get("after")
-        before = kwargs.get("before")
-        if not (sessionID or after or before):
-            raise ValueError("sessionID or after or before must be defined")
+        validate_feeds_required_parameters(kwargs)
+        endpoint = kwargs.pop("endpoint", Endpoint.FEED.value)
+        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint)
+
+        return self._results(
+            f"domain-registration-data-access-protocol-feed-({source.value})",
+            f"v1/{endpoint}/domainrdap/",
+            response_path=(),
+            **kwargs,
+        )
+
+    def domaindiscovery(self, **kwargs):
+        """Returns new domains as they are either discovered in domain registration information, observed by our global sensor network, or reported by trusted third parties"""
+        validate_feeds_required_parameters(kwargs)
+        endpoint = kwargs.pop("endpoint", Endpoint.FEED.value)
+        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint)
+        if endpoint == Endpoint.DOWNLOAD.value or kwargs.get("output_format", OutputFormat.JSONL.value) != OutputFormat.CSV.value:
+            kwargs.pop("headers", None)
 
         return self._results(
-            "domain-registration-data-access-protocol-feed-(api)",
-            "v1/feed/domainrdap/",
+            f"real-time-domain-discovery-feed-({source.value})",
+            f"v1/{endpoint}/domaindiscovery/",
             response_path=(),
             **kwargs,
         )

domaintools/base_results.py

@@ -4,8 +4,12 @@
 import re
 import time
 import logging
+
+from copy import deepcopy
 from datetime import datetime
+from httpx import Client
 
+from domaintools.constants import OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
 from domaintools.exceptions import (
     BadRequestException,
     InternalServerErrorException,
@@ -18,7 +22,6 @@
 )
 from domaintools.utils import get_feeds_products_list
 
-from httpx import Client
 
 try:  # pragma: no cover
     from collections.abc import MutableMapping, MutableSequence
@@ -90,6 +93,18 @@ def _make_request(self):
                 patch_data = self.kwargs.copy()
                 patch_data.update(self.api.extra_request_params)
                 return session.patch(url=self.url, json=patch_data)
+            elif self.product in get_feeds_products_list():
+                parameters = deepcopy(self.kwargs)
+                parameters.pop("output_format", None)
+                parameters.pop(
+                    "format", None
+                )  # For some unknownn reasons, even if "format" is not included in the cli params for feeds endpoint, it is being populated thus we need to remove it. Happens only if using CLI.
+                headers = {}
+                if self.kwargs.get("output_format", OutputFormat.JSONL.value) == OutputFormat.CSV.value:
+                    parameters["headers"] = int(bool(self.kwargs.get("headers", False)))
+                    headers["accept"] = HEADER_ACCEPT_KEY_CSV_FORMAT
+
+                return session.get(url=self.url, params=parameters, headers=headers, **self.api.extra_request_params)
             else:
                 return session.get(url=self.url, params=self.kwargs, **self.api.extra_request_params)
 
@@ -259,6 +274,32 @@ def json(self):
             **self.kwargs,
         )
 
+    @property
+    def jsonl(self):
+        self.kwargs.pop("format", None)
+        return self.__class__(
+            format="jsonl",
+            product=self.product,
+            url=self.url,
+            items_path=self.items_path,
+            response_path=self.response_path,
+            api=self.api,
+            **self.kwargs,
+        )
+
+    @property
+    def csv(self):
+        self.kwargs.pop("format", None)
+        return self.__class__(
+            format="csv",
+            product=self.product,
+            url=self.url,
+            items_path=self.items_path,
+            response_path=self.response_path,
+            api=self.api,
+            **self.kwargs,
+        )
+
     @property
     def xml(self):
         self.kwargs.pop("format", None)

domaintools/constants.py

@@ -0,0 +1,24 @@
+from enum import Enum
+
+
+class Endpoint(Enum):
+    FEED = "feed"
+    DOWNLOAD = "download"
+
+
+class Source(Enum):
+    API = "api"
+    S3 = "s3"
+
+
+class OutputFormat(Enum):
+    JSONL = "jsonl"
+    CSV = "csv"
+
+
+HEADER_ACCEPT_KEY_CSV_FORMAT = "text/csv"
+
+ENDPOINT_TO_SOURCE_MAP = {
+    Endpoint.FEED.value: Source.API,
+    Endpoint.DOWNLOAD.value: Source.S3,
+}

domaintools/utils.py

@@ -176,4 +176,15 @@ def get_feeds_products_list():
         "newly-active-domains-feed-(api)",
         "newly-observed-domains-feed-(api)",
         "domain-registration-data-access-protocol-feed-(api)",
+        "domain-registration-data-access-protocol-feed-(s3)",
+        "real-time-domain-discovery-feed-(api)",
+        "real-time-domain-discovery-feed-(s3)",
     ]
+
+
+def validate_feeds_required_parameters(params):
+    sessionID = params.get("sessionID")
+    after = params.get("after")
+    before = params.get("before")
+    if not (sessionID or after or before):
+        raise ValueError("sessionID or after or before must be defined")

domaintools_async/__init__.py

@@ -1,11 +1,14 @@
 """Adds async capabilities to the base product object"""
 
 import asyncio
+
+from copy import deepcopy
 from httpx import AsyncClient
 
 from domaintools.base_results import Results
-
-from domaintools.exceptions import ServiceUnavailableException, ServiceException
+from domaintools.constants import OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
+from domaintools.exceptions import ServiceUnavailableException
+from domaintools.utils import get_feeds_products_list
 
 
 class _AIter(object):
@@ -49,6 +52,17 @@ async def _make_async_request(self, session):
             patch_data = self.kwargs.copy()
             patch_data.update(self.api.extra_request_params)
             results = await session.patch(url=self.url, json=patch_data)
+        elif self.product in get_feeds_products_list():
+            parameters = deepcopy(self.kwargs)
+            parameters.pop("output_format", None)
+            parameters.pop(
+                "format", None
+            )  # For some unknownn reasons, even if "format" is not included in the cli params for feeds endpoint, it is being populated thus we need to remove it. Happens only if using CLI.
+            headers = {}
+            if self.kwargs.get("output_format", OutputFormat.JSONL.value) == OutputFormat.CSV.value:
+                parameters["headers"] = int(bool(self.kwargs.get("headers", False)))
+                headers["accept"] = HEADER_ACCEPT_KEY_CSV_FORMAT
+            results = await session.get(url=self.url, params=parameters, headers=headers, **self.api.extra_request_params)
         else:
             results = await session.get(url=self.url, params=self.kwargs, **self.api.extra_request_params)
         if results: