[IDEV-959] enhancement: use dt python wrapper in feeds integration

Packs/FeedDomainTools/Integrations/FeedDomainTools/FeedDomainTools.py

@@ -4,15 +4,15 @@
 
 from CommonServerPython import *  # noqa: F401
 from collections.abc import Callable, Iterator
-
+from domaintools import API
 
 # disable insecure warnings
 urllib3.disable_warnings()
 
 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,16 +125,14 @@ 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")
         domain = dt_feed_kwargs.get("domain")
         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)}")

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"

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.