Add phishstats analyzer

api_app/analyzers_manager/observable_analyzers/phishstats.py

@@ -13,45 +13,74 @@
 
 class PhishStats(ObservableAnalyzer):
     """
-    Analyzer that uses PhishStats API to check if the observable is a phishing site.
+    Analyzer that uses PhishStats API to check if the observable
+    is a phishing site.
     """
 
-    url: str = "https://api.phishstats.info/api"
+    url: str = "https://api.phishstats.info/api/phishing"
 
     @classmethod
     def update(cls) -> bool:
-        pass
+        return True
 
-    # Gate on a real listing hit: the $malicious mapping constant would otherwise
-    # stamp MALICIOUS on every clean lookup.
+    # Mark the observable as malicious only when PhishStats returns
+    # an actual matching record.
     def _do_create_data_model(self) -> bool:
         return super()._do_create_data_model() and bool(self.report.report.get("results"))
 
-    def __build_phishstats_url(self) -> str:
-        to_analyze_observable_classification = self.observable_classification
-        to_analyze_observable_name = self.observable_name
-        if self.observable_classification == Classification.URL:
-            to_analyze_observable_name = urlparse(self.observable_name).hostname
+    def __build_phishstats_url(self) -> tuple[str, dict]:
+        observable_classification = self.observable_classification
+        observable_name = self.observable_name
+
+        if observable_classification == Classification.URL:
+            parsed_url = urlparse(observable_name)
+            observable_name = parsed_url.hostname
+
+            if not observable_name or not parsed_url.scheme:
+                raise AnalyzerRunException("Invalid URL observable")
+
             try:
-                IPv4Address(to_analyze_observable_name)
+                IPv4Address(observable_name)
             except AddressValueError:
-                to_analyze_observable_classification = Classification.DOMAIN
+                observable_classification = Classification.DOMAIN
             else:
-                to_analyze_observable_classification = Classification.IP
-
-        if to_analyze_observable_classification == Classification.IP:
-            endpoint = f"phishing?_where=(ip,eq,{to_analyze_observable_name})&_sort=-date"
-        elif to_analyze_observable_classification == Classification.DOMAIN:
-            endpoint = f"phishing?_where=(url,like,~{to_analyze_observable_name}~)&_sort=-date"
-        elif to_analyze_observable_classification == Classification.GENERIC:
-            endpoint = f"phishing?_where=(title,like,~{to_analyze_observable_name}~)&_sort=-date"
+                observable_classification = Classification.IP
+
+        if not observable_name:
+            raise AnalyzerRunException("Invalid PhishStats observable")
+
+        if observable_classification == Classification.IP:
+            where = f"(ip,eq,{observable_name})"
+
+        elif observable_classification == Classification.DOMAIN:
+            where = f"(url,like,~{observable_name}~)"
+
+        elif observable_classification == Classification.GENERIC:
+            where = f"(title,like,~{observable_name}~)"
+
         else:
-            raise AnalyzerRunException("Phishstats require either of IP, URL, Domain or Generic")
-        return f"{self.url}/{endpoint}"
+            raise AnalyzerRunException("PhishStats requires an IP, URL, Domain or Generic observable")
+
+        params = {
+            "_where": where,
+            "_sort": "-date",
+        }
+
+        return self.url, params
 
     def run(self):
-        api_url = self.__build_phishstats_url()
-        response = requests.get(api_url)
+        api_url, params = self.__build_phishstats_url()
+
+        response = requests.get(
+            api_url,
+            params=params,
+            timeout=30,
+        )
         response.raise_for_status()
 
-        return {"api_url": api_url, "results": response.json()}
+        results = response.json()
+
+        return {
+            "api_url": response.url,
+            "results": results,
+        }

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_phishstats.py

@@ -1,6 +1,8 @@
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
+from api_app.analyzers_manager.exceptions import AnalyzerRunException
 from api_app.analyzers_manager.observable_analyzers.phishstats import PhishStats
+from api_app.choices import Classification
 from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
     BaseAnalyzerTest,
 )
@@ -26,5 +28,54 @@ def get_mocked_response():
         ]
         return patch(
             "requests.get",
-            return_value=MockUpResponse(mock_json_response, 200),
+            return_value=MockUpResponse(
+                mock_json_response,
+                200,
+                url="https://api.phishstats.info/api/phishing",
+            ),
         )
+
+    def _make_analyzer(self, observable_type, observable_value):
+        analyzer = PhishStats.__new__(PhishStats)
+        analyzer.observable_classification = observable_type
+        analyzer.observable_name = observable_value
+        return analyzer
+
+    @patch("api_app.analyzers_manager.observable_analyzers.phishstats.requests.get")
+    def test_url_observable_uses_domain_filter(self, mock_get):
+        mock_get.return_value = MagicMock()
+        mock_get.return_value.url = "https://api.phishstats.info/api/phishing"
+        mock_get.return_value.json.return_value = []
+
+        analyzer = self._make_analyzer(Classification.URL, "https://example.com/login")
+        result = analyzer.run()
+
+        mock_get.assert_called_once_with(
+            "https://api.phishstats.info/api/phishing",
+            params={
+                "_where": "(url,like,~example.com~)",
+                "_sort": "-date",
+            },
+            timeout=30,
+        )
+        self.assertEqual(result["results"], [])
+
+    @patch("api_app.analyzers_manager.observable_analyzers.phishstats.requests.get")
+    def test_ip_url_uses_ip_filter(self, mock_get):
+        mock_get.return_value = MagicMock()
+        mock_get.return_value.url = "https://api.phishstats.info/api/phishing"
+        mock_get.return_value.json.return_value = []
+
+        analyzer = self._make_analyzer(Classification.URL, "https://8.8.8.8/login")
+        analyzer.run()
+
+        self.assertEqual(
+            mock_get.call_args.kwargs["params"]["_where"],
+            "(ip,eq,8.8.8.8)",
+        )
+
+    def test_invalid_url_raises_exception(self):
+        analyzer = self._make_analyzer(Classification.URL, "not-a-valid-url")
+
+        with self.assertRaises(AnalyzerRunException):
+            analyzer.run()