Vulnerable code sample

import json
import requests

class VulnerableHttpToolPlugin:
    """
    A simplified class representing the vulnerable logic in the pre-1.1.3
    utcp-http plugin, demonstrating CVE-2024-4461.
    """
    def __init__(self):
        # This dictionary simulates the agent's internal tool registry.
        self._tools = {}

    def _is_discovery_url_allowed(self, url):
        # Simulates the initial validation, which only allows HTTPS.
        # An attacker can bypass this by hosting their spec on any HTTPS server.
        return url.startswith("https://")

    def _fetch_spec_from_url(self, url):
        # Simulates fetching a malicious OpenAPI spec from an attacker's server
        # that is hosted on an allowed (HTTPS) domain.
        spec_text = """
        {
            "openapi": "3.0.0",
            "info": { "title": "Malicious Tool", "version": "1.0.0" },
            "servers": [
                { "url": "http://127.0.0.1:9090/internal-status" }
            ],
            "paths": {
                "/get_info": {
                    "get": { "operationId": "get_internal_info" }
                }
            }
        }
        """
        return json.loads(spec_text)

    def register_manual(self, discovery_url: str):
        """
        Vulnerable registration method. It validates the discovery_url
        but blindly trusts the `servers` URL within the fetched spec.
        """
        if not self._is_discovery_url_allowed(discovery_url):
            raise ValueError("Discovery URL must use HTTPS.")

        spec = self._fetch_spec_from_url(discovery_url)

        # VULNERABILITY: The URL from the spec's `servers` field is
        # extracted and trusted without being validated against the allowlist.
        unvalidated_tool_url = spec["servers"][0]["url"]
        tool_name = spec["paths"]["/get_info"]["get"]["operationId"]

        # The tool is registered with the unvalidated, potentially malicious URL.
        self._tools[tool_name] = {"url": unvalidated_tool_url}

    def call_tool(self, tool_name: str):
        """
        Vulnerable tool-calling method. It reuses the stored URL without
        re-validating it, leading to a Server-Side Request Forgery (SSRF).
        """
        tool_info = self._tools.get(tool_name)
        if not tool_info:
            raise NameError(f"Tool '{tool_name}' not found.")

        target_url = tool_info["url"]

        # VULNERABILITY: The unvalidated URL is used to make an HTTP request.
        # This request is sent to the internal address (`http://127.0.0.1:9090`)
        # defined by the attacker in the OpenAPI spec. There is no re-validation.
        response = requests.get(target_url)
        return response.text