diff --git a/src/vuln_analysis/register.py b/src/vuln_analysis/register.py index 0b339ff7a..066f12aed 100644 --- a/src/vuln_analysis/register.py +++ b/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 diff --git a/src/vuln_analysis/utils/clients/intel_client.py b/src/vuln_analysis/utils/clients/intel_client.py index f2e936f69..bc5970438 100644 --- a/src/vuln_analysis/utils/clients/intel_client.py +++ b/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 diff --git a/src/vuln_analysis/utils/clients/nvd_client.py b/src/vuln_analysis/utils/clients/nvd_client.py index fd6a43d54..e7987156f 100644 --- a/src/vuln_analysis/utils/clients/nvd_client.py +++ b/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") diff --git a/src/vuln_analysis/utils/http_utils.py b/src/vuln_analysis/utils/http_utils.py index f58f87ea6..bcc4f2a68 100644 --- a/src/vuln_analysis/utils/http_utils.py +++ b/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 diff --git a/src/vuln_analysis/utils/serp_api_wrapper.py b/src/vuln_analysis/utils/serp_api_wrapper.py index f2f559b83..763a4c035 100644 --- a/src/vuln_analysis/utils/serp_api_wrapper.py +++ b/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 diff --git a/src/vuln_analysis/utils/url_utils.py b/src/vuln_analysis/utils/url_utils.py index 07e43725e..2a54d6384 100644 --- a/src/vuln_analysis/utils/url_utils.py +++ b/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..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 diff --git a/tests/test_url_utils.py b/tests/test_url_utils.py new file mode 100644 index 000000000..c23284ff1 --- /dev/null +++ b/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