IDEV-2272 and IDEV-2273: support header auth for iris endpoints and change default Feeds auth behavior

README.md

@@ -215,25 +215,25 @@ Please see the [supported versions](https://github.com/DomainTools/python_api/ra
 for the DomainTools Python support policy.
 
 
-Real-Time Threat Intelligence Feeds
+Real-Time Threat Feeds
 ===================
 
-Real-Time Threat Intelligence Feeds provide data on the different stages of the domain lifecycle: from first-observed in the wild, to newly re-activated after a period of quiet. Access current feed data in real-time or retrieve historical feed data through separate APIs.
+Real-Time Threat Feeds provide data on the different stages of the domain lifecycle: from first-observed in the wild, to newly re-activated after a period of quiet. Access current feed data in real-time or retrieve historical feed data through separate APIs.
 
 Custom parameters aside from the common `GET` Request parameters:
 - `endpoint` (choose either `download` or `feed` API endpoint - default is `feed`)
     ```python
-    api = API(USERNAME, KEY, always_sign_api_key=False)
+    api = API(USERNAME, KEY)
     api.nod(endpoint="feed", **kwargs)
     ```
 - `header_authentication`: by default, we're using API Header Authentication. Set this False if you want to use API Key and Secret Authentication. Apparently, you can't use API Header Authentication for `download` endpoints so this will be defaulted to `False` even without explicitly setting it.
     ```python
-    api = API(USERNAME, KEY, always_sign_api_key=False)
-    api.nod(header_authentication=False, **kwargs)
+    api = API(USERNAME, KEY, header_authentication=False)
+    api.nod(**kwargs)
     ```
 - `output_format`: (choose either `csv` or `jsonl` - default is `jsonl`). Cannot be used in `domainrdap` feeds. Additionally, `csv` is not available for `download` endpoints.
     ```python
-    api = API(USERNAME, KEY, always_sign_api_key=False)
+    api = API(USERNAME, KEY)
     api.nod(output_format="csv", **kwargs)
     ```
 
@@ -254,7 +254,7 @@ Since we may dealing with large feeds datasets, the python wrapper uses `generat
 ```python
 from domaintools import API
 
-api = API(USERNAME, KEY, always_sign_api_key=False)
+api = API(USERNAME, KEY)
 results = api.nod(sessionID="my-session-id", after=-60)
 
 for result in results.response() # generator that holds NOD feeds data for the past 60 seconds and is expected to request only once
@@ -265,7 +265,7 @@ for result in results.response() # generator that holds NOD feeds data for the p
 ```python
 from domaintools import API
 
-api = API(USERNAME, KEY, always_sign_api_key=False)
+api = API(USERNAME, KEY)
 results = api.nod(sessionID="my-session-id", after=-7200)
 
 for partial_result in results.response() # generator that holds NOD feeds data for the past 2 hours and is expected to request multiple times

domaintools/api.py

@@ -6,7 +6,7 @@
 import re
 import ssl
 
-from domaintools.constants import Endpoint, ENDPOINT_TO_SOURCE_MAP, FEEDS_PRODUCTS_LIST, OutputFormat
+from domaintools.constants import Endpoint, ENDPOINT_TO_SOURCE_MAP, RTTF_PRODUCTS_LIST, OutputFormat
 from domaintools._version import current as version
 from domaintools.results import (
     GroupedIterable,
@@ -63,7 +63,8 @@ def __init__(
         verify_ssl=True,
         rate_limit=True,
         proxy_url=None,
-        always_sign_api_key=True,
+        always_sign_api_key=None,
+        header_authentication=None,
         key_sign_hash="sha256",
         app_name="python_wrapper",
         app_version=version,
@@ -83,6 +84,7 @@ def __init__(
         self.proxy_url = proxy_url
         self.extra_request_params = {}
         self.always_sign_api_key = always_sign_api_key
+        self.header_authentication = header_authentication
         self.key_sign_hash = key_sign_hash
         self.default_parameters["app_name"] = app_name
         self.default_parameters["app_version"] = app_version
@@ -129,19 +131,27 @@ 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
-        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)
+        is_rttf_product = product in RTTF_PRODUCTS_LIST
+        self._handle_api_key_parameters(is_rttf_product)
+        self.handle_api_key(is_rttf_product, path, parameters)
         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, product, path, parameters, header_authentication):
+    def _handle_api_key_parameters(self, is_rttf_product):
+        if self.always_sign_api_key is None:
+            self.always_sign_api_key = not is_rttf_product
+
+        if self.header_authentication is None:
+            self.header_authentication = is_rttf_product
+
+    def handle_api_key(self, is_rttf_product, path, parameters):
         if self.https and not self.always_sign_api_key:
-            if product in FEEDS_PRODUCTS_LIST and header_authentication:
-                parameters["X-Api-Key"] = self.key
-            else:
-                parameters["api_key"] = self.key
+            parameters["api_key"] = self.key
         else:
+            if is_rttf_product:
+                # As per requirement in IDEV-2272, raise this error when the user explicitly sets signing of API key for RTTF endpoints
+                raise ValueError("Real Time Threat Feeds do not support signed API keys.")
             if self.key_sign_hash and self.key_sign_hash in AVAILABLE_KEY_SIGN_HASHES:
                 signing_hash = eval(self.key_sign_hash)
             else:

domaintools/base_results.py

@@ -9,7 +9,7 @@
 from datetime import datetime
 from httpx import Client
 
-from domaintools.constants import FEEDS_PRODUCTS_LIST, OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
+from domaintools.constants import RTTF_PRODUCTS_LIST, OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
 from domaintools.exceptions import (
     BadRequestException,
     InternalServerErrorException,
@@ -75,45 +75,45 @@ def _wait_time(self):
 
         return wait_for
 
-    def _get_session_params(self):
-        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.
+    def _get_session_params_and_headers(self):
         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
-
-        header_api_key = parameters.pop("X-Api-Key", None)
-        if header_api_key:
-            headers["X-Api-Key"] = header_api_key
+        parameters = deepcopy(self.kwargs)
+        is_rttf_product = self.product in RTTF_PRODUCTS_LIST
+        if is_rttf_product:
+            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.
+            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
+
+        if self.api.header_authentication:
+            header_key_for_api_key = "X-Api-Key" if is_rttf_product else "X-API-Key"
+            headers[header_key_for_api_key] = parameters.pop("api_key", None)
 
         return {"parameters": parameters, "headers": headers}
 
     def _make_request(self):
 
         with Client(verify=self.api.verify_ssl, proxy=self.api.proxy_url, timeout=None) as session:
+            session_params_and_headers = self._get_session_params_and_headers()
+            headers = session_params_and_headers.get("headers")
             if self.product in [
                 "iris-investigate",
                 "iris-enrich",
                 "iris-detect-escalate-domains",
             ]:
                 post_data = self.kwargs.copy()
                 post_data.update(self.api.extra_request_params)
-                return session.post(url=self.url, data=post_data)
+                return session.post(url=self.url, data=post_data, headers=headers)
             elif self.product in ["iris-detect-manage-watchlist-domains"]:
                 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 FEEDS_PRODUCTS_LIST:
-                session_params = self._get_session_params()
-                parameters = session_params.get("parameters")
-                headers = session_params.get("headers")
-                return session.get(url=self.url, params=parameters, headers=headers, **self.api.extra_request_params)
+                return session.patch(url=self.url, json=patch_data, headers=headers)
             else:
-                return session.get(url=self.url, params=self.kwargs, **self.api.extra_request_params)
+                parameters = session_params_and_headers.get("parameters")
+                return session.get(url=self.url, params=parameters, headers=headers, **self.api.extra_request_params)
 
     def _get_results(self):
         wait_for = self._wait_time()
@@ -170,7 +170,7 @@ def status(self):
 
     def setStatus(self, code, response=None):
         self._status = code
-        if code == 200 or (self.product in FEEDS_PRODUCTS_LIST and code == 206):
+        if code == 200 or (self.product in RTTF_PRODUCTS_LIST and code == 206):
             return
 
         reason = None

domaintools/cli/api.py

@@ -9,7 +9,7 @@
 from rich.progress import Progress, SpinnerColumn, TextColumn
 
 from domaintools.api import API
-from domaintools.constants import Endpoint, FEEDS_PRODUCTS_LIST, OutputFormat
+from domaintools.constants import Endpoint, RTTF_PRODUCTS_LIST, OutputFormat
 from domaintools.cli.utils import get_file_extension
 from domaintools.exceptions import ServiceException
 from domaintools._version import current as version
@@ -110,7 +110,7 @@ def args_to_dict(*args) -> Dict:
     def _get_formatted_output(cls, cmd_name: str, response, out_format: str = "json"):
         if cmd_name in ("available_api_calls",):
             return "\n".join(response)
-        if response.product in FEEDS_PRODUCTS_LIST:
+        if response.product in RTTF_PRODUCTS_LIST:
             return "\n".join([data for data in response.response()])
         return str(getattr(response, out_format) if out_format != "list" else response.as_list())
 
@@ -180,6 +180,7 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
             out_file = params.pop("out_file", sys.stdout)
             verify_ssl = params.pop("no_verify_ssl", False)
             always_sign_api_key = params.pop("no_sign_api_key", False)
+            header_authentication = params.pop("no_header_authentication", False)
             source = None
 
             if "src_file" in params:
@@ -214,6 +215,7 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
                     verify_ssl=verify_ssl,
                     rate_limit=rate_limit,
                     always_sign_api_key=always_sign_api_key,
+                    header_authentication=header_authentication,
                 )
                 dt_api_func = getattr(dt_api, name)
 
@@ -229,7 +231,7 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
 
                 if isinstance(out_file, _io.TextIOWrapper):
                     # use rich `print` command to prettify the ouput in sys.stdout
-                    if response.product in FEEDS_PRODUCTS_LIST:
+                    if response.product in RTTF_PRODUCTS_LIST:
                         print(output)
                     else:
                         print(response)

domaintools/cli/commands/feeds.py

@@ -32,8 +32,8 @@ def feeds_nad(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),
@@ -112,8 +112,8 @@ def feeds_nod(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),
@@ -192,8 +192,8 @@ def feeds_domainrdap(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),
@@ -260,8 +260,8 @@ def feeds_domaindiscovery(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),
@@ -340,8 +340,8 @@ def feeds_noh(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),
@@ -420,8 +420,8 @@ def feeds_domainhotlist(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),
@@ -500,8 +500,8 @@ def feeds_realtime_domain_risk(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
-    header_authentication: bool = typer.Option(
-        True,
+    no_header_authentication: bool = typer.Option(
+        False,
         "--no-header-auth",
         help="Don't use header authentication",
     ),