Merge pull request #207 from DomainTools/IDEV-2586

[IDEV-2586]: feat: add HMAC-SHA256 signing support for RTTF feed endpoints

Author: briluza

README.md

@@ -269,6 +269,12 @@ Custom parameters aside from the common `GET` Request parameters:
     api = API(USERNAME, KEY, header_authentication=False)
     api.nod(**kwargs)
     ```
+- `always_sign_api_key`: set to `True` to use HMAC-SHA256 signed authentication instead of header auth. When set, `header_authentication` automatically defaults to `False` — both methods do not fire simultaneously. The signing algorithm is identical to the standard API: `HMAC-SHA256(key, username + timestamp + path)`, with `timestamp` and `signature` sent as query parameters.
+    ```python
+    api = API(USERNAME, KEY, always_sign_api_key=True)
+    api.nod(after="-60")
+    # sends: api_username, timestamp, signature — no X-Api-Key header
+    ```
 - `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)

domaintools/api.py

@@ -196,17 +196,16 @@ def _handle_api_key_parameters(self, is_rttf_product):
             self.always_sign_api_key = not is_rttf_product
 
         if self.header_authentication is None:
-            self.header_authentication = is_rttf_product
+            # When HMAC signing is explicitly requested for RTTF, disable header auth
+            # so both methods don't fire simultaneously
+            self.header_authentication = is_rttf_product and not self.always_sign_api_key
 
     def handle_api_key(self, is_rttf_product, path, parameters):
         if self.header_authentication and not self.always_sign_api_key:
             return
         if self.https and not self.always_sign_api_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:
@@ -215,10 +214,12 @@ def handle_api_key(self, is_rttf_product, path, parameters):
                     "Values available are {1}".format(self.key_sign_hash, ",".join(AVAILABLE_KEY_SIGN_HASHES))
                 )
 
+            # RTTF paths lack a leading slash; normalize before signing
+            sign_path = path if path.startswith("/") else f"/{path}"
             parameters["timestamp"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
             parameters["signature"] = hmac(
                 self.key.encode("utf8"),
-                "".join([self.username, parameters["timestamp"], path]).encode("utf8"),
+                "".join([self.username, parameters["timestamp"], sign_path]).encode("utf8"),
                 digestmod=signing_hash,
             ).hexdigest()
 

examples/rttf_feeds.py

@@ -0,0 +1,30 @@
+from domaintools import API
+
+api = API(USER_NAME, KEY)
+
+# --- Stream feed (default) ---
+# Streams newly observed domains from the last 60 seconds.
+results = api.nod(after="-60", top=10)
+for line in results.response():
+    print(line)
+
+# --- Stream feed with HMAC signing ---
+# Uses HMAC-SHA256 instead of the default X-Api-Key header.
+# header_authentication is automatically disabled when always_sign_api_key=True.
+hmac_api = API(USER_NAME, KEY, always_sign_api_key=True)
+results = hmac_api.nod(after="-60", top=10)
+for line in results.response():
+    print(line)
+
+# --- Stream feed with sessionID ---
+# Each subsequent call returns only data since the last request.
+results = api.nod(sessionID="my-session", after="-3600", top=10)
+for line in results.response():
+    print(line)
+
+# --- Download endpoint ---
+# Returns a JSON listing of available S3 batch files (not a stream).
+result = api.nod(endpoint="download", limit=5)
+print(result["download_name"])
+for f in result["files"]:
+    print(f["name"], f["url"])

tests/test_api.py

@@ -895,16 +895,56 @@ def test_ip_risk():
         assert "all_threats_combined_percent" in feed_result.keys()
 
 
-@vcr.use_cassette
-def test_feeds_endpoint_should_raise_error_if_signed_api_key_is_used():
-    feeds_api.always_sign_api_key = True
-    try:
-        with pytest.raises(ValueError) as excinfo:
-            feeds_api.domaindiscovery(after="-60")
-        assert str(excinfo.value) == "Real Time Threat Feeds do not support signed API keys."
-    finally:
-        feeds_api.always_sign_api_key = False
-        feeds_api.header_authentication = True
+def test_rttf_hmac_produces_timestamp_and_signature_not_api_key():
+    """RTTF with always_sign_api_key=True must add timestamp+signature and omit api_key."""
+    hmac_api = API("testuser", "testkey", rate_limit=False, always_sign_api_key=True)
+    result = hmac_api.nod(after="-60")
+    session_info = result._get_session_params_and_headers()
+    params = session_info["parameters"]
+
+    assert "timestamp" in params
+    assert "signature" in params
+    assert "api_key" not in params
+    assert "X-Api-Key" not in session_info["headers"]
+
+
+def test_rttf_hmac_auto_disables_header_authentication():
+    """When always_sign_api_key=True, header_authentication must default to False for RTTF."""
+    hmac_api = API("testuser", "testkey", rate_limit=False, always_sign_api_key=True)
+    hmac_api.nod(after="-60")
+    assert hmac_api.header_authentication is False
+
+
+def test_rttf_hmac_signature_is_correct():
+    """RTTF HMAC signature must match manual calculation using the normalised /v1/feed/... path."""
+    from hashlib import sha256
+    from hmac import new as hmac_new
+
+    hmac_api = API("testuser", "testkey", rate_limit=False, always_sign_api_key=True)
+    result = hmac_api.nod(after="-60")
+    params = result._get_session_params_and_headers()["parameters"]
+
+    ts = params["timestamp"]
+    expected = hmac_new(
+        "testkey".encode("utf8"),
+        f"testuser{ts}/v1/feed/nod/".encode("utf8"),
+        digestmod=sha256,
+    ).hexdigest()
+    assert params["signature"] == expected
+
+
+def test_rttf_hmac_explicit_header_auth_false_still_signs():
+    """Explicit header_authentication=False with always_sign_api_key=True must produce a signature."""
+    hmac_api = API(
+        "testuser", "testkey",
+        rate_limit=False,
+        always_sign_api_key=True,
+        header_authentication=False,
+    )
+    result = hmac_api.nod(after="-60")
+    params = result._get_session_params_and_headers()["parameters"]
+    assert "signature" in params
+    assert "api_key" not in params
 
 
 def test_rttf_api_key_not_leaked_as_query_param():