IDEV-2011: Implement support for API header authentication for Real-Time Threat Intelligence Feeds. Add download endpoint support for nod and nad.

Author: jbabac

domaintools/api.py

@@ -19,7 +19,7 @@
     filter_by_field,
     DTResultFilter,
 )
-from domaintools.utils import validate_feeds_parameters
+from domaintools.utils import get_feeds_products_list, validate_feeds_parameters
 
 
 AVAILABLE_KEY_SIGN_HASHES = ["sha1", "sha256", "md5"]
@@ -125,14 +125,18 @@ def _results(self, product, path, cls=Results, **kwargs):
         uri = "/".join((self._rest_api_url, path.lstrip("/")))
         parameters = self.default_parameters.copy()
         parameters["api_username"] = self.username
-        self.handle_api_key(path, parameters)
+        header_authentication = kwargs.pop("header_authentication", True)  # Used only by Real-Time Threat Intelligence Feeds endpoints for now
+        self.handle_api_key(product, path, parameters, header_authentication)
         parameters.update({key: str(value).lower() if value in (True, False) else value for key, value in kwargs.items() if value is not None})
 
         return cls(self, product, uri, **parameters)
 
-    def handle_api_key(self, path, parameters):
+    def handle_api_key(self, product, path, parameters, header_authentication):
         if self.https and not self.always_sign_api_key:
-            parameters["api_key"] = self.key
+            if product in get_feeds_products_list() and header_authentication:
+                parameters["X-Api-Key"] = self.key
+            else:
+                parameters["api_key"] = self.key
         else:
             if self.key_sign_hash and self.key_sign_hash in AVAILABLE_KEY_SIGN_HASHES:
                 signing_hash = eval(self.key_sign_hash)
@@ -1063,28 +1067,32 @@ def iris_detect_ignored_domains(
 
     def nod(self, **kwargs):
         """Returns back list of the newly observed domains feed"""
-        sessionID = kwargs.get("sessionID")
-        after = kwargs.get("after")
-        if not (sessionID or after):
-            raise ValueError("sessionID or after (can be both) must be defined")
+        validate_feeds_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:
+            # headers param is allowed only in Feed API and CSV format
+            kwargs.pop("headers", None)
 
         return self._results(
-            "newly-observed-domains-feed-(api)",
-            "v1/feed/nod/",
+            f"newly-observed-domains-feed-({source.value})",
+            f"v1/{endpoint}/nod/",
             response_path=(),
             **kwargs,
         )
 
     def nad(self, **kwargs):
         """Returns back list of the newly active domains feed"""
-        sessionID = kwargs.get("sessionID")
-        after = kwargs.get("after")
-        if not (sessionID or after):
-            raise ValueError("sessionID or after (can be both) must be defined")
+        validate_feeds_parameters(kwargs)
+        endpoint = kwargs.pop("endpoint", Endpoint.FEED.value)
+        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint).value
+        if endpoint == Endpoint.DOWNLOAD.value or kwargs.get("output_format", OutputFormat.JSONL.value) != OutputFormat.CSV.value:
+            # headers param is allowed only in Feed API and CSV format
+            kwargs.pop("headers", None)
 
         return self._results(
-            "newly-active-domains-feed-(api)",
-            "v1/feed/nad/",
+            f"newly-active-domains-feed-({source})",
+            f"v1/{endpoint}/nad/",
             response_path=(),
             **kwargs,
         )
@@ -1093,10 +1101,10 @@ def domainrdap(self, **kwargs):
         """Returns changes to global domain registration information, populated by the Registration Data Access Protocol (RDAP)"""
         validate_feeds_parameters(kwargs)
         endpoint = kwargs.pop("endpoint", Endpoint.FEED.value)
-        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint)
+        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint).value
 
         return self._results(
-            f"domain-registration-data-access-protocol-feed-({source.value})",
+            f"domain-registration-data-access-protocol-feed-({source})",
             f"v1/{endpoint}/domainrdap/",
             response_path=(),
             **kwargs,
@@ -1106,13 +1114,13 @@ 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_parameters(kwargs)
         endpoint = kwargs.pop("endpoint", Endpoint.FEED.value)
-        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint)
+        source = ENDPOINT_TO_SOURCE_MAP.get(endpoint).value
         if endpoint == Endpoint.DOWNLOAD.value or kwargs.get("output_format", OutputFormat.JSONL.value) != OutputFormat.CSV.value:
             # headers param is allowed only in Feed API and CSV format
             kwargs.pop("headers", None)
 
         return self._results(
-            f"real-time-domain-discovery-feed-({source.value})",
+            f"real-time-domain-discovery-feed-({source})",
             f"v1/{endpoint}/domaindiscovery/",
             response_path=(),
             **kwargs,

domaintools/base_results.py

@@ -104,6 +104,10 @@ def _make_request(self):
                     parameters["headers"] = int(bool(self.kwargs.get("headers", False)))
                     headers["accept"] = HEADER_ACCEPT_KEY_CSV_FORMAT
 
+                header_api_key = parameters.pop("X-Api-Key", None)
+                if header_api_key:
+                    headers["X-Api-Key"] = header_api_key
+
                 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)

domaintools_async/__init__.py

@@ -62,6 +62,11 @@ async def _make_async_request(self, session):
             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
+
+            header_api_key = parameters.pop("X-Api-Key", None)
+            if header_api_key:
+                headers["X-Api-Key"] = header_api_key
+
             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)