From 9573e59d56573da6173be28cb354f9e1073428df Mon Sep 17 00:00:00 2001 From: bluza Date: Mon, 13 Jul 2026 22:21:10 +0800 Subject: [PATCH] use dt python wrapper in feeds integration --- .../FeedDomainTools/FeedDomainTools.py | 91 +++++++++++-------- .../FeedDomainTools/FeedDomainTools_test.py | 63 +++++++++++++ Packs/FeedDomainTools/README.md | 2 +- 3 files changed, 115 insertions(+), 41 deletions(-) diff --git a/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools.py b/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools.py index e6ef429cd435..b4544140ec3b 100644 --- a/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools.py +++ b/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools.py @@ -4,7 +4,7 @@ from CommonServerPython import * # noqa: F401 from collections.abc import Callable, Iterator - +from domaintools import API # disable insecure warnings urllib3.disable_warnings() @@ -12,7 +12,7 @@ RISK_THRESHOLD = 70 -class DomainToolsClient(BaseClient): +class DomainToolsClient: """ Client to use in the DomainTools Feed integration. """ @@ -29,8 +29,15 @@ class DomainToolsClient(BaseClient): DOMAINRISK = "domainrisk" DOMAINHOTLIST = "domainhotlist" - FEED_URL = "/v1/feed" - DOMAINTOOLS_API_BASE_URL = "https://api.domaintools.com" + FEED_METHOD_MAP = { + "nod": "nod", + "nad": "nad", + "noh": "noh", + "domainrdap": "domainrdap", + "domaindiscovery": "domaindiscovery", + "domainrisk": "realtime_domain_risk", + "domainhotlist": "domainhotlist", + } def __init__( self, @@ -41,50 +48,55 @@ def __init__( tags: str = "", tlp_color: str | None = None, ): - self.feed_type = "nod" # default to NOD feeds - self.tags = tags - self.tlp_color = tlp_color - if not (api_username and api_key): raise DemistoException("The 'API Username' and 'API Key' parameters are required.") - self.api_username = api_username - self.api_key = api_key + self.tags = tags + self.tlp_color = tlp_color - super().__init__( - base_url=self.DOMAINTOOLS_API_BASE_URL, headers={"Content-Type": "application/json"}, verify=verify_ssl, proxy=proxy + proxy_url = None + if proxy: + proxies = handle_proxy() + proxy_url = proxies.get("https") or proxies.get("http") or None + + self._api = API( + api_username, + api_key, + app_partner=self.APP_PARTNER, + app_name=self.APP_NAME, + app_version=self.APP_VERSION, + proxy_url=proxy_url, + verify_ssl=verify_ssl, + always_sign_api_key=True, ) def _get_dt_feeds( self, + feed_type: str, session_id: str | None = None, domain: str | None = None, after: str | None = None, before: str | None = None, top: int | None = None, ) -> list[str]: - feed_type_name = self.feed_type.upper() - - query_params = { - "api_key": self.api_key, - "api_username": self.api_username, - "app_partner": self.APP_PARTNER, - "app_name": self.APP_NAME, - "top": top, - "sessionID": session_id, - "after": after, - "before": before, - "domain": domain, + method_name = self.FEED_METHOD_MAP[feed_type] + api_method = getattr(self._api, method_name) + + kwargs: dict[str, Any] = { + k: v + for k, v in { + "sessionID": session_id, + "domain": domain, + "after": after, + "before": before, + "top": top, + }.items() + if v is not None } - demisto.info(f"Fetching DomainTools {feed_type_name} feed type with params: {query_params}") + demisto.info(f"Fetching DomainTools {feed_type.upper()} feed type with params: {kwargs}") - response = self._http_request( - "GET", url_suffix=f"{self.FEED_URL}/{self.feed_type}/", params=query_params, resp_type="text", raise_on_status=True - ) - - results = response.strip().split("\n") if response else [] - return results + return list(api_method(**kwargs).response()) def _format_parameter(self, key: str, value: Any) -> Any: """Format the parameter value based on the given key @@ -113,8 +125,6 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I Returns: list: A list of objects, containing the indicators. """ - self.feed_type = feed_type - # DomainTools feeds optional arguments session_id = dt_feed_kwargs.get("session_id", "dt-cortex-feeds") top = int(dt_feed_kwargs.get("top") or "5000") @@ -122,7 +132,7 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I after = dt_feed_kwargs.get("after") before = dt_feed_kwargs.get("before") - demisto.info(f"Start building list of indicators for {self.feed_type} feed.") + demisto.info(f"Start building list of indicators for {feed_type} feed.") limit_counter = 0 processed_feeds = 0 @@ -136,6 +146,7 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I before = self._format_parameter(key="before", value=before) dt_feeds = self._get_dt_feeds( + feed_type=feed_type, session_id=session_id, domain=domain, after=after, @@ -144,7 +155,7 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I ) total_dt_feeds = len(dt_feeds) - demisto.info(f"Fetched {total_dt_feeds} of {self.feed_type} feeds.") + demisto.info(f"Fetched {total_dt_feeds} of {feed_type} feeds.") ud_tags = [tag.strip() for tag in self.tags.split(",")] @@ -167,14 +178,14 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I "value": indicator, "type": indicator_type, "timestamp": timestamp, - "tags": ["DomainToolsFeeds", self.feed_type] + ud_tags, + "tags": ["DomainToolsFeeds", feed_type] + ud_tags, "tlp_color": self.tlp_color, "parsed_record": parsed_record, "overall_risk_score": overall_risk_score, } - # for domaihotlist & domainriks feed, we will be returning the risk scores - if self.feed_type in (self.DOMAINRISK, self.DOMAINHOTLIST): + # for domainhotlist & domainrisk feed, we will be returning the risk scores + if feed_type in (self.DOMAINRISK, self.DOMAINHOTLIST): risk_score_details = { "phishing_risk": json_feed.get("phishing_risk"), "malware_risk": json_feed.get("malware_risk"), @@ -183,7 +194,7 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I "overall_risk": json_feed.get("overall_risk"), } - if self.feed_type == self.DOMAINHOTLIST: + if feed_type == self.DOMAINHOTLIST: risk_score_details["expires"] = json_feed.get("expires") # update the parsed dt feed data @@ -195,7 +206,7 @@ def build_iterator(self, feed_type: str = "nod", dt_feed_kwargs: dict = {}) -> I limit_counter += 1 processed_feeds += 1 - demisto.info(f"Done processing {processed_feeds} out of {total_dt_feeds} {self.feed_type} feeds.") + demisto.info(f"Done processing {processed_feeds} out of {total_dt_feeds} {feed_type} feeds.") except Exception as err: demisto.debug(str(err)) raise ValueError(f"Could not parse returned data as indicator. \n\nError massage: {str(err)}") diff --git a/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools_test.py b/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools_test.py index 84ccb36cd749..36349aabb1b9 100644 --- a/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools_test.py +++ b/Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools_test.py @@ -234,3 +234,66 @@ def test_calling_command_using_main(mocker, dt_feeds_client): main() results = demisto.results.call_args[0] assert results[0] == "ok" + + +class TestGetDtFeeds: + def test_dispatches_correct_api_method(self, mocker, dt_feeds_client): + """_get_dt_feeds calls the correct domaintools API method via FEED_METHOD_MAP.""" + mock_response = mocker.MagicMock() + mock_response.response.return_value = iter([]) + mock_method = mocker.MagicMock(return_value=mock_response) + mocker.patch.object(dt_feeds_client._api, "realtime_domain_risk", mock_method, create=True) + + dt_feeds_client._get_dt_feeds(feed_type="domainrisk", top=10) + + mock_method.assert_called_once_with(top=10) + + def test_filters_none_kwargs(self, mocker, dt_feeds_client): + """None values are not passed as kwargs to the API method.""" + mock_response = mocker.MagicMock() + mock_response.response.return_value = iter([]) + mock_method = mocker.MagicMock(return_value=mock_response) + mocker.patch.object(dt_feeds_client._api, "nod", mock_method, create=True) + + dt_feeds_client._get_dt_feeds(feed_type="nod", session_id="s1", domain=None, top=5) + + call_kwargs = mock_method.call_args.kwargs + assert "domain" not in call_kwargs + assert call_kwargs == {"sessionID": "s1", "top": 5} + + def test_returns_list_of_lines(self, mocker, dt_feeds_client): + """Returns list of NDJSON lines from FeedsResults.response().""" + lines = ['{"domain":"example.com"}', '{"domain":"test.com"}'] + mock_response = mocker.MagicMock() + mock_response.response.return_value = iter(lines) + mocker.patch.object(dt_feeds_client._api, "nod", return_value=mock_response, create=True) + + result = dt_feeds_client._get_dt_feeds(feed_type="nod") + + assert result == lines + + +def test_missing_credentials(): + """DomainToolsClient raises DemistoException when credentials are empty.""" + with pytest.raises(DemistoException): + DomainToolsClient(api_username="", api_key="") + + +def test_format_parameter_prepends_dash(dt_feeds_client): + """_format_parameter prepends '-' to after/before values that lack it.""" + assert dt_feeds_client._format_parameter("after", "60") == "-60" + assert dt_feeds_client._format_parameter("after", "-60") == "-60" + assert dt_feeds_client._format_parameter("before", "120") == "-120" + + +def test_test_module_all_feed_type_falls_back_to_nod(mocker, dt_feeds_client): + """test_module falls back to 'nod' when feed_type param is 'ALL'.""" + from FeedDomainTools import test_module + + mocker.patch.object( + dt_feeds_client, + "_get_dt_feeds", + return_value=feed_mock_response.NOD_FEED_RESPONSE, + ) + result = test_module(dt_feeds_client, args={}, params={"feed_type": "ALL"}) + assert result == "ok" diff --git a/Packs/FeedDomainTools/README.md b/Packs/FeedDomainTools/README.md index d8460a6b0c14..634097521616 100644 --- a/Packs/FeedDomainTools/README.md +++ b/Packs/FeedDomainTools/README.md @@ -1,2 +1,2 @@ The DomainTools Feed provides real-time access to newly registered and observed domains, enabling proactive threat detection and defense. -This integration was integrated and tested with version 1.0.4 of FeedDomainTools. +This integration was integrated and tested with version 1.1.0 of FeedDomainTools.