IDEV-2272: Change default feeds auth behavior.

Author: jbabac

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,20 +131,31 @@ 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:
+            if self.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:
+            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.")
+            elif self.key_sign_hash and self.key_sign_hash in AVAILABLE_KEY_SIGN_HASHES:
                 signing_hash = eval(self.key_sign_hash)
             else:
                 raise ValueError(

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,
@@ -107,7 +107,7 @@ 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 FEEDS_PRODUCTS_LIST:
+            elif self.product in RTTF_PRODUCTS_LIST:
                 session_params = self._get_session_params()
                 parameters = session_params.get("parameters")
                 headers = session_params.get("headers")
@@ -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())
 
@@ -229,7 +229,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

@@ -27,16 +27,6 @@ def feeds_nad(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     output_format: str = typer.Option(
         "jsonl",
         "-f",
@@ -107,16 +97,6 @@ def feeds_nod(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     output_format: str = typer.Option(
         "jsonl",
         "-f",
@@ -187,16 +167,6 @@ def feeds_domainrdap(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     endpoint: str = typer.Option(
         Endpoint.FEED.value,
         "-e",
@@ -255,16 +225,6 @@ def feeds_domaindiscovery(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     output_format: str = typer.Option(
         "jsonl",
         "-f",
@@ -335,16 +295,6 @@ def feeds_noh(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     output_format: str = typer.Option(
         "jsonl",
         "-f",
@@ -415,16 +365,6 @@ def feeds_domainhotlist(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     output_format: str = typer.Option(
         "jsonl",
         "-f",
@@ -495,16 +435,6 @@ def feeds_realtime_domain_risk(
         "--no-verify-ssl",
         help="Skip verification of SSL certificate when making HTTPs API calls",
     ),
-    no_sign_api_key: bool = typer.Option(
-        False,
-        "--no-sign-api-key",
-        help="Skip signing of api key",
-    ),
-    header_authentication: bool = typer.Option(
-        True,
-        "--no-header-auth",
-        help="Don't use header authentication",
-    ),
     output_format: str = typer.Option(
         "jsonl",
         "-f",

domaintools/constants.py

@@ -23,7 +23,7 @@ class OutputFormat(Enum):
     Endpoint.DOWNLOAD.value: Source.S3,
 }
 
-FEEDS_PRODUCTS_LIST = [
+RTTF_PRODUCTS_LIST = [
     "newly-active-domains-feed-(api)",
     "newly-active-domains-feed-(s3)",
     "newly-observed-domains-feed-(api)",

domaintools/utils.py

@@ -183,7 +183,3 @@ def validate_feeds_parameters(params):
     endpoint = params.get("endpoint")
     if endpoint == Endpoint.DOWNLOAD.value and format == OutputFormat.CSV.value:
         raise ValueError(f"{format} format is not available in {Endpoint.DOWNLOAD.value} API.")
-
-    if endpoint == Endpoint.DOWNLOAD.value and params.get("header_authentication", True):
-        # For download endpoint, header_authentication will be False by default
-        params["header_authentication"] = False

domaintools_async/__init__.py

@@ -6,7 +6,7 @@
 from httpx import AsyncClient
 
 from domaintools.base_results import Results
-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 ServiceUnavailableException
 
 
@@ -51,7 +51,7 @@ 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 FEEDS_PRODUCTS_LIST:
+        elif self.product in RTTF_PRODUCTS_LIST:
             session_params = self._get_session_params()
             parameters = session_params.get("parameters")
             headers = session_params.get("headers")