fix: allowlist configurable intel base URLs to prevent SSRF (APPENG-5712 / T-010)

src/vuln_analysis/register.py

@@ -65,6 +65,7 @@
 # pylint: enable=unused-import
 from vuln_analysis.utils.llm_engine_utils import postprocess_engine_output, finalize_preprocess_engine_input
 from vuln_analysis.utils.llm_engine_utils import preprocess_engine_input
+from vuln_analysis.utils.url_utils import validate_all_base_urls
 from exploit_iq_commons.logging.loggers_factory import LoggingFactory, trace_id
 
 logger = LoggingFactory.get_agent_logger(__name__)
@@ -127,7 +128,7 @@ class CVEAgentWorkflowConfig(FunctionBaseConfig, name="cve_agent"):
 
 @register_function(config_type=CVEAgentWorkflowConfig, framework_wrappers=[LLMFrameworkEnum.LANGCHAIN])
 async def cve_agent_workflow(config: CVEAgentWorkflowConfig, builder: Builder):
-
+    validate_all_base_urls()
     from langgraph.graph import END
     from langgraph.graph import START
     from langgraph.graph import StateGraph

src/vuln_analysis/utils/clients/intel_client.py

@@ -19,6 +19,7 @@
 import aiohttp
 
 from vuln_analysis.utils.async_http_utils import request_with_retry
+from vuln_analysis.utils.url_utils import validate_base_url
 
 
 class IntelClient:
@@ -38,6 +39,8 @@ def __init__(self,
         if (base_url is None):
             base_url = self.default_base_url()
 
+        # SSRF allowlist check for configurable base URLs.
+        validate_base_url(base_url)
         self._session = session
 
         self._base_url = base_url

src/vuln_analysis/utils/clients/nvd_client.py

@@ -25,6 +25,7 @@
 from vuln_analysis.utils.intel_utils import parse
 from vuln_analysis.utils.intel_utils import parse_config_vendors
 from vuln_analysis.utils.url_utils import url_join
+from vuln_analysis.utils.url_utils import validate_base_url
 from vuln_analysis.utils.clients.intel_client import IntelClient
 
 from exploit_iq_commons.logging.loggers_factory import LoggingFactory
@@ -62,7 +63,10 @@ def __init__(self,
 
         self._api_key = api_key or os.environ.get('NVD_API_KEY', None)
 
-        self._cwe_details_url_template = url_join(os.environ.get('CWE_DETAILS_BASE_URL', self.CWE_DETAILS_URL),
+        cwe_details_base_url = os.environ.get('CWE_DETAILS_BASE_URL', self.CWE_DETAILS_URL)
+        # SSRF allowlist check.
+        validate_base_url(cwe_details_base_url)
+        self._cwe_details_url_template = url_join(cwe_details_base_url,
                                                   "data/definitions",
                                                   "{CWE_ID}.html")
 

src/vuln_analysis/utils/http_utils.py

@@ -19,7 +19,6 @@
 from http import HTTPStatus
 
 import requests
-import urllib3
 
 from exploit_iq_commons.logging.loggers_factory import LoggingFactory
 logger = LoggingFactory.get_agent_logger(__name__)
@@ -108,20 +107,3 @@ def request_with_retry(
             logger.debug("Sleeping for %s seconds before retrying request again", actual_sleep_time)
             time.sleep(actual_sleep_time)
 
-
-def prepare_url(url: str) -> str:
-    """
-    Verifies that `url` contains a protocol scheme and a host and returns the url.
-    If no protocol scheme is provided, `http` is used.
-    """
-    parsed_url = urllib3.util.parse_url(url)
-    if parsed_url.scheme is None or parsed_url.host is None:
-        url = f"http://{url}"
-
-        parsed_url = urllib3.util.parse_url(url)
-        if parsed_url.scheme is None or parsed_url.host is None:
-            raise ValueError(f"Invalid URL: {url}")
-
-        logger.warning("No protocol scheme provided in URL, using: %s", url)
-
-    return parsed_url.url

src/vuln_analysis/utils/serp_api_wrapper.py

@@ -24,6 +24,7 @@
 from vuln_analysis.utils.async_http_utils import retry_async
 from vuln_analysis.utils.sanitization import wrap_untrusted_data
 from vuln_analysis.utils.url_utils import url_join
+from vuln_analysis.utils.url_utils import validate_base_url as validate_base_url_allowlist
 
 
 class ExploitIqSerpAPIWrapper(SerpAPIWrapper):
@@ -63,6 +64,8 @@ def validate_base_url(self) -> "ExploitIqSerpAPIWrapper":
         self.base_url = get_from_env(key="base_url", env_key="SERPAPI_BASE_URL", default=self.base_url)
         if not self.base_url:
             raise ValueError("SERPAPI_BASE_URL must not be empty")
+        # SSRF allowlist check.
+        validate_base_url_allowlist(self.base_url)
         # Update the base URL for search_engine
         self.search_engine.BACKEND = self.base_url
         return self

src/vuln_analysis/utils/url_utils.py

@@ -12,6 +12,12 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import os
+from urllib.parse import urlparse
+
+from exploit_iq_commons.logging.loggers_factory import LoggingFactory
+
+logger = LoggingFactory.get_agent_logger(__name__)
 
 
 def url_join(*parts):
@@ -25,3 +31,103 @@ def url_join(*parts):
         The joined URL.
     """
     return "/".join(part.strip("/") for part in parts)
+
+
+_ALLOWED_BASE_URL_HOSTS = frozenset({
+    # Internal caching proxy (docker-compose / kustomize deployment design)
+    "nginx-cache",
+    "localhost",
+    # Genuine external service hosts (each client's default_base_url())
+    "api.github.com",            # GHSA
+    "services.nvd.nist.gov",     # NVD
+    "api.first.org",             # FIRST / EPSS
+    "access.redhat.com",         # RHSA
+    "osidb.prodsec.redhat.com",  # OSIDB
+    "ubuntu.com",                # Ubuntu
+    "api.deps.dev",              # deps.dev
+    "serpapi.com",               # SerpAPI
+    "cwe.mitre.org",             # CWE details
+})
+
+_ALLOWED_BASE_URL_SCHEMES = frozenset({"http", "https"})
+
+# Env vars that configure external service base URLs (CTRL-007 / T-010).
+_BASE_URL_ENV_VARS = (
+    "NVD_BASE_URL",
+    "GHSA_BASE_URL",
+    "RHSA_BASE_URL",
+    "FIRST_BASE_URL",
+    "UBUNTU_BASE_URL",
+    "OSIDB_BASE_URL",
+    "DEPSDEV_BASE_URL",
+    "SERPAPI_BASE_URL",
+    "CWE_DETAILS_BASE_URL",
+)
+
+
+def _is_allowed_base_url_host(hostname: str | None) -> bool:
+    """Return True if hostname is an allowed base-URL target (CTRL-007 / T-010)."""
+    if not hostname:
+        return False
+    if hostname in _ALLOWED_BASE_URL_HOSTS:
+        return True
+    # In-cluster FQDN of the nginx cache: exactly
+    # nginx-cache.<namespace>.svc.cluster.local (5 dot-separated labels).
+    # Deliberately NOT a blanket "*.svc.cluster.local" nor deeper subdomains --
+    # that would open the entire cluster DNS namespace.
+    parts = hostname.split(".")
+    if len(parts) == 5 and parts[0] == "nginx-cache" and parts[2:] == ["svc", "cluster", "local"]:
+        return True
+    return False
+
+
+def validate_base_url(url: str) -> str:
+    """Validate a configurable external-service base URL (CTRL-007 / T-010).
+
+    Ensures the URL uses an allowed scheme (http/https) and that its hostname is
+    in the hardcoded allowlist, preventing SSRF redirection of intel requests to
+    arbitrary internal services (e.g. cloud metadata, the k8s API).
+
+    Raises:
+        ValueError: If the URL is empty, uses a disallowed scheme, or its host
+                    is not in the allowlist.
+    """
+    if not url:
+        raise ValueError("Base URL is empty")
+
+    parsed = urlparse(url)
+
+    if parsed.scheme not in _ALLOWED_BASE_URL_SCHEMES:
+        raise ValueError(
+            f"Base URL scheme must be http or https, got: {parsed.scheme!r}. "
+            f"Rejected URL: {url}"
+        )
+
+    if not _is_allowed_base_url_host(parsed.hostname):
+        raise ValueError(
+            f"Base URL host not in allowlist: {parsed.hostname!r}. "
+            f"Rejected URL: {url}"
+        )
+
+    logger.debug("Base URL validated: %s", url)
+    return url
+
+
+def validate_all_base_urls() -> None:
+    """Validate every configured external-service base URL at startup (CTRL-007 / T-010).
+
+    Reads each ``*_BASE_URL`` env var and validates any that are set. Unset vars
+    are skipped -- clients fall back to their hardcoded (trusted) default_base_url().
+    Raises on the first invalid value so misconfiguration fails fast at startup
+    rather than at first use.
+
+    Raises:
+        ValueError: If any configured base URL fails validation.
+    """
+    for env_var in _BASE_URL_ENV_VARS:
+        value = os.environ.get(env_var)
+        if value:
+            try:
+                validate_base_url(value)
+            except ValueError as e:
+                raise ValueError(f"Invalid {env_var}: {e}") from e

tests/test_url_utils.py

@@ -0,0 +1,68 @@
+import pytest
+
+from vuln_analysis.utils.url_utils import validate_base_url
+from vuln_analysis.utils.url_utils import _BASE_URL_ENV_VARS
+from vuln_analysis.utils.clients.nvd_client import NVDClient
+from vuln_analysis.utils.clients.ghsa_client import GHSAClient
+from vuln_analysis.utils.clients.rhsa_client import RHSAClient
+from vuln_analysis.utils.clients.first_client import FirstClient
+from vuln_analysis.utils.clients.ubuntu_client import UbuntuClient
+from vuln_analysis.utils.clients.osidb_client import OsidbClient
+from vuln_analysis.utils.url_utils import validate_all_base_urls
+from vuln_analysis.utils.vulnerable_dependency_checker import VulnerableDependencyChecker
+
+
+@pytest.mark.parametrize("url", [
+    "http://nginx-cache/nvd",                                              # kustomize short name
+    "http://nginx-cache.exploit-iq-tests.svc.cluster.local:8080/nvd",     # Tekton FQDN
+    "http://localhost:8080/serpapi",                                       # README local dev
+    "https://serpapi.com",
+    "https://services.nvd.nist.gov/rest",
+    "http://cwe.mitre.org",
+])
+def test_allows_known_hosts(url):
+    assert validate_base_url(url) == url
+
+
+@pytest.mark.parametrize("url", [
+    "http://169.254.169.254/latest/meta-data/",          # cloud metadata
+    "http://kubernetes.default.svc/",                    # k8s API
+    "http://evil.svc.cluster.local:8080/nvd",           # other in-cluster svc (must NOT pass)
+    "http://nginx-cache.a.b.svc.cluster.local/nvd",     # deeper subdomain -> must NOT pass
+    "http://127.0.0.1:4318/",                           # loopback IP (not "localhost") -> rejected
+    "http://evil.example.com/",
+    "file:///etc/passwd",
+    "",
+])
+def test_rejects_disallowed(url):
+    with pytest.raises(ValueError):
+        validate_base_url(url)
+
+
+# Regression guard: IntelClient validates even the default base URL, so a new
+# client whose default host is missing from the allowlist would fail to start.
+@pytest.mark.parametrize("cls", [
+    NVDClient, GHSAClient, RHSAClient, FirstClient,
+    UbuntuClient, OsidbClient, VulnerableDependencyChecker,
+])
+def test_every_client_default_base_url_is_allowlisted(cls):
+    validate_base_url(cls.default_base_url())
+
+
+def _clear_base_url_env(monkeypatch):
+    """Isolate from any *_BASE_URL already set in the shell/CI environment."""
+    for env_var in _BASE_URL_ENV_VARS:
+        monkeypatch.delenv(env_var, raising=False)
+
+
+def test_validate_all_base_urls_rejects_bad_env(monkeypatch):
+    _clear_base_url_env(monkeypatch)
+    monkeypatch.setenv("NVD_BASE_URL", "http://169.254.169.254/nvd")
+    with pytest.raises(ValueError):
+        validate_all_base_urls()
+
+
+def test_validate_all_base_urls_allows_good_env(monkeypatch):
+    _clear_base_url_env(monkeypatch)
+    monkeypatch.setenv("NVD_BASE_URL", "http://nginx-cache/nvd")
+    validate_all_base_urls()  # should not raise