diff --git a/README.md b/README.md index 7a83a9c..dffb3de 100644 --- a/README.md +++ b/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) diff --git a/domaintools/api.py b/domaintools/api.py index 581dce9..c5b965c 100644 --- a/domaintools/api.py +++ b/domaintools/api.py @@ -196,7 +196,9 @@ 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: @@ -204,9 +206,6 @@ def handle_api_key(self, is_rttf_product, path, parameters): 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() diff --git a/examples/rttf_feeds.py b/examples/rttf_feeds.py new file mode 100644 index 0000000..a6d6cb7 --- /dev/null +++ b/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"]) diff --git a/tests/test_api.py b/tests/test_api.py index 867928f..cbdff16 100644 --- a/tests/test_api.py +++ b/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():