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

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():