Validate user-supplied URLs in tool API clients

dojo/fixtures/unit_sonarqube_toolConfig1.json

@@ -5,7 +5,7 @@
     "fields": {
       "name": "SQ1",
       "description": null,
-      "url": "http://localhost/",
+      "url": "https://8.8.8.8/",
       "tool_type": 1,
       "authentication_type": "API",
       "extras": null,

dojo/fixtures/unit_sonarqube_toolConfig2.json

@@ -5,7 +5,7 @@
     "fields": {
       "name": "SQ2",
       "description": null,
-      "url": "http://localhost/",
+      "url": "https://8.8.8.8/",
       "tool_type": 1,
       "authentication_type": "API",
       "extras": null,

dojo/tools/api_bugcrowd/api_client.py

@@ -1,8 +1,9 @@
 from urllib.parse import urlencode
 
-import requests
 from django.conf import settings
 
+from dojo.utils_ssrf import make_ssrf_safe_session
+
 
 class BugcrowdAPI:
 
@@ -16,7 +17,7 @@ class BugcrowdAPI:
     }
 
     def __init__(self, tool_config):
-        self.session = requests.Session()
+        self.session = make_ssrf_safe_session()
         if tool_config.authentication_type == "API":
             self.api_token = tool_config.api_key
             self.session.headers.update(

dojo/tools/api_cobalt/api_client.py

@@ -1,6 +1,7 @@
-import requests
 from django.conf import settings
 
+from dojo.utils_ssrf import make_ssrf_safe_session
+
 
 class CobaltAPI:
 
@@ -9,7 +10,7 @@ class CobaltAPI:
     cobalt_api_url = "https://api.cobalt.io"
 
     def __init__(self, tool_config):
-        self.session = requests.Session()
+        self.session = make_ssrf_safe_session()
         if tool_config.authentication_type == "API":
             self.api_token = tool_config.api_key
             self.org_token = tool_config.extras

dojo/tools/api_edgescan/api_client.py

@@ -1,9 +1,10 @@
 import json
 from json.decoder import JSONDecodeError
 
-import requests
 from django.conf import settings
 
+from dojo.utils_ssrf import SSRFError, make_ssrf_safe_session, validate_url_for_ssrf
+
 
 class EdgescanAPI:
 
@@ -15,6 +16,12 @@ def __init__(self, tool_config):
         if tool_config.authentication_type == "API":
             self.api_key = tool_config.api_key
             self.url = tool_config.url or self.DEFAULT_URL
+            try:
+                validate_url_for_ssrf(self.url)
+            except SSRFError as e:
+                msg = f"Edgescan URL is not allowed: {e}"
+                raise ValueError(msg) from e
+            self.session = make_ssrf_safe_session()
             self.options = self.get_extra_options(tool_config)
         else:
             msg = f"Edgescan Authentication type {tool_config.authentication_type} not supported"
@@ -39,7 +46,7 @@ def get_findings(self, asset_ids):
         if self.options and "date" in self.options:
             url += f"&c[date_opened_after]={self.options['date']}"
 
-        response = requests.get(
+        response = self.session.get(
             url=url,
             headers=self.get_headers(),
             proxies=self.get_proxies(),

dojo/tools/api_sonarqube/api_client.py

@@ -1,12 +1,18 @@
-import requests
 from django.conf import settings
 from requests.exceptions import JSONDecodeError as RequestsJSONDecodeError
 
 from dojo.utils import prepare_for_view
+from dojo.utils_ssrf import SSRFError, make_ssrf_safe_session, validate_url_for_ssrf
 
 
 class SonarQubeAPI:
     def __init__(self, tool_config):
+        try:
+            validate_url_for_ssrf(tool_config.url)
+        except SSRFError as e:
+            msg = f"SonarQube URL is not allowed: {e}"
+            raise ValueError(msg) from e
+
         self.rules_cache = {}
 
         supported_issue_types = ["BUG", "VULNERABILITY", "CODE_SMELL"]
@@ -42,7 +48,7 @@ def __init__(self, tool_config):
                 msg = f"Detected unsupported issue type! Supported types are {', '.join(supported_issue_types)}"
                 raise Exception(msg)
 
-        self.session = requests.Session()
+        self.session = make_ssrf_safe_session()
         self.default_headers = {"User-Agent": "DefectDojo"}
         self.sonar_api_url = tool_config.url
         if tool_config.authentication_type == "Password":

dojo/tools/api_vulners/api_client.py

@@ -1,5 +1,7 @@
 import vulners
 
+from dojo.utils_ssrf import SSRFError, validate_url_for_ssrf
+
 
 class VulnersAPI:
 
@@ -12,6 +14,11 @@ def __init__(self, tool_config):
         if tool_config.authentication_type == "API":
             self.api_key = tool_config.api_key
             if tool_config.url:
+                try:
+                    validate_url_for_ssrf(tool_config.url)
+                except SSRFError as e:
+                    msg = f"Vulners URL is not allowed: {e}"
+                    raise ValueError(msg) from e
                 self.vulners_api_url = tool_config.url
         else:
             msg = f"Vulners.com Authentication type {tool_config.authentication_type} not supported"

unittests/test_tool_api_clients_ssrf.py

@@ -0,0 +1,134 @@
+import socket
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from dojo.tools.api_bugcrowd.api_client import BugcrowdAPI
+from dojo.tools.api_cobalt.api_client import CobaltAPI
+from dojo.tools.api_edgescan.api_client import EdgescanAPI
+from dojo.tools.api_sonarqube.api_client import SonarQubeAPI
+from dojo.tools.api_vulners.api_client import VulnersAPI
+from dojo.utils_ssrf import _SSRFSafeAdapter  # noqa: PLC2701
+from unittests.dojo_test_case import DojoTestCase
+
+
+def _sonarqube_config(url):
+    return SimpleNamespace(
+        url=url,
+        authentication_type="API",
+        api_key="dummy-key",
+        extras=None,
+    )
+
+
+def _edgescan_config(url):
+    return SimpleNamespace(
+        url=url,
+        authentication_type="API",
+        api_key="dummy-key",
+        extras=None,
+    )
+
+
+def _vulners_config(url):
+    return SimpleNamespace(
+        url=url,
+        authentication_type="API",
+        api_key="dummy-key",
+    )
+
+
+def _bugcrowd_config():
+    return SimpleNamespace(
+        authentication_type="API",
+        api_key="dummy-key",
+    )
+
+
+def _cobalt_config():
+    return SimpleNamespace(
+        authentication_type="API",
+        api_key="dummy-key",
+        extras=None,
+    )
+
+
+class TestSonarQubeUrlValidation(DojoTestCase):
+
+    def test_private_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "SonarQube URL is not allowed"):
+            SonarQubeAPI(_sonarqube_config("http://192.168.1.1/"))
+
+    def test_loopback_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "SonarQube URL is not allowed"):
+            SonarQubeAPI(_sonarqube_config("http://127.0.0.1/"))
+
+    def test_link_local_metadata_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "SonarQube URL is not allowed"):
+            SonarQubeAPI(_sonarqube_config("http://169.254.169.254/"))
+
+    def test_public_url_succeeds(self):
+        # 8.8.8.8 is a numeric literal — no DNS lookup required, so this is
+        # reliable in CI.
+        client = SonarQubeAPI(_sonarqube_config("http://8.8.8.8/"))
+        self.assertEqual(client.sonar_api_url, "http://8.8.8.8/")
+
+
+class TestEdgescanUrlValidation(DojoTestCase):
+
+    def test_private_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "Edgescan URL is not allowed"):
+            EdgescanAPI(_edgescan_config("http://192.168.1.1/"))
+
+    def test_loopback_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "Edgescan URL is not allowed"):
+            EdgescanAPI(_edgescan_config("http://127.0.0.1/"))
+
+    def test_public_url_succeeds(self):
+        client = EdgescanAPI(_edgescan_config("http://8.8.8.8/"))
+        self.assertEqual(client.url, "http://8.8.8.8/")
+
+    def test_default_url_succeeds(self):
+        # tool_config.url=None should fall back to DEFAULT_URL (a public host).
+        with patch("dojo.utils_ssrf.socket.getaddrinfo") as mock_getaddrinfo:
+            mock_getaddrinfo.return_value = [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80)),
+            ]
+            client = EdgescanAPI(_edgescan_config(None))
+            self.assertEqual(client.url, EdgescanAPI.DEFAULT_URL)
+
+
+class TestVulnersUrlValidation(DojoTestCase):
+
+    def test_private_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "Vulners URL is not allowed"):
+            VulnersAPI(_vulners_config("http://192.168.1.1/"))
+
+    def test_loopback_url_raises(self):
+        with self.assertRaisesRegex(ValueError, "Vulners URL is not allowed"):
+            VulnersAPI(_vulners_config("http://127.0.0.1/"))
+
+    def test_public_url_succeeds(self):
+        client = VulnersAPI(_vulners_config("http://8.8.8.8/"))
+        self.assertEqual(client.vulners_api_url, "http://8.8.8.8/")
+
+    def test_no_url_uses_library_default(self):
+        # When no URL is configured, the validation is skipped and the
+        # external `vulners` library uses its own default endpoint.
+        client = VulnersAPI(_vulners_config(None))
+        self.assertIsNone(client.vulners_api_url)
+
+
+class TestBugcrowdSessionIsSafe(DojoTestCase):
+
+    def test_session_uses_ssrf_safe_adapter(self):
+        client = BugcrowdAPI(_bugcrowd_config())
+        for adapter in client.session.adapters.values():
+            self.assertIsInstance(adapter, _SSRFSafeAdapter)
+
+
+class TestCobaltSessionIsSafe(DojoTestCase):
+
+    def test_session_uses_ssrf_safe_adapter(self):
+        client = CobaltAPI(_cobalt_config())
+        for adapter in client.session.adapters.values():
+            self.assertIsInstance(adapter, _SSRFSafeAdapter)

unittests/tools/test_api_sonarqube_importer.py

@@ -292,7 +292,7 @@ def test_parser(self):
         self.assertEqual('Remove this useless assignment to local variable "currentValue".', finding.title)
         self.assertEqual(None, finding.cwe)
         self.assertEqual("", finding.description)
-        self.assertEqual("[Issue permalink](http://localhoproject/issues?issues=AWKWIl8pZpu0CyehMfc4&open=AWKWIl8pZpu0CyehMfc4&resolved=CONFIRMED&id=internal.dummy.project) \n", finding.references)
+        self.assertEqual("[Issue permalink](https://8.8.8project/issues?issues=AWKWIl8pZpu0CyehMfc4&open=AWKWIl8pZpu0CyehMfc4&resolved=CONFIRMED&id=internal.dummy.project) \n", finding.references)
         self.assertEqual("Medium", finding.severity)
         self.assertEqual(242, finding.line)
         self.assertEqual("internal.dummy.project:src/main/javascript/TranslateDirective.ts", finding.file_path)
@@ -516,7 +516,7 @@ def test_parser(self):
         )
         self.assertEqual(str(findings[0].severity), "High")
         self.assertMultiLineEqual(
-            "[Hotspot permalink](http://localhosecurity_hotspots?id=internal.dummy.project&hotspots=AXgm6Z-ophPPY0C1qhRq) "
+            "[Hotspot permalink](https://8.8.8security_hotspots?id=internal.dummy.project&hotspots=AXgm6Z-ophPPY0C1qhRq) "
             "\n"
             "[CVE-2019-13466](http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2019-13466)"
             "\n"
@@ -586,7 +586,7 @@ def test_parser(self):
             findings[0].description,
         )
         self.assertEqual(str(findings[0].severity), "High")
-        self.assertEqual(findings[0].references, "[Hotspot permalink](http://localhosecurity_hotspots?id=internal.dummy.project&hotspots=AXgm6Z-ophPPY0C1qhRq) \n")
+        self.assertEqual(findings[0].references, "[Hotspot permalink](https://8.8.8security_hotspots?id=internal.dummy.project&hotspots=AXgm6Z-ophPPY0C1qhRq) \n")
         self.assertEqual(str(findings[0].file_path), "internal.dummy.project:spec/support/user_fixture.rb")
         self.assertEqual(findings[0].line, 9)
         self.assertEqual(findings[0].active, True)

unittests/tools/test_api_sonarqube_parser.py

@@ -46,7 +46,7 @@ def setUp(self):
         # build Sonarqube conf (the parser need it)
         tool_type, _ = Tool_Type.objects.get_or_create(name="SonarQube")
         tool_conf, _ = Tool_Configuration.objects.get_or_create(
-            name="SQ1_unittests", authentication_type="API", tool_type=tool_type, url="http://dummy.url.foo.bar/api",
+            name="SQ1_unittests", authentication_type="API", tool_type=tool_type, url="https://8.8.8.8/api",
         )
         pasc, _ = Product_API_Scan_Configuration.objects.get_or_create(
             product=product, tool_configuration=tool_conf, service_key_1="ABCD",