diff --git "a/src/backend/base/langflow/initial_setup/starter_projects/Pok\303\251dex Agent.json" "b/src/backend/base/langflow/initial_setup/starter_projects/Pok\303\251dex Agent.json" index 40eb687ae9c3..b1a7d112c2a0 100644 --- "a/src/backend/base/langflow/initial_setup/starter_projects/Pok\303\251dex Agent.json" +++ "b/src/backend/base/langflow/initial_setup/starter_projects/Pok\303\251dex Agent.json" @@ -771,7 +771,7 @@ "key": "APIRequest", "legacy": false, "metadata": { - "code_hash": "4f329dec9a39", + "code_hash": "1a052ebb9519", "dependencies": { "dependencies": [ { @@ -889,7 +889,7 @@ "show": true, "title_case": false, "type": "code", - "value": "import json\nimport re\nimport tempfile\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom typing import Any\nfrom urllib.parse import parse_qsl, urlencode, urlparse, urlunparse\n\nimport aiofiles\nimport aiofiles.os as aiofiles_os\nimport httpx\nimport validators\n\nfrom lfx.base.curl.parse import parse_context\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import TabInput\nfrom lfx.io import (\n BoolInput,\n DataInput,\n DropdownInput,\n IntInput,\n MessageTextInput,\n MultilineInput,\n Output,\n TableInput,\n)\nfrom lfx.schema.data import Data\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.utils.component_utils import set_current_fields, set_field_advanced, set_field_display\n\n# SSRF Protection imports - for preventing Server-Side Request Forgery attacks\nfrom lfx.utils.ssrf_protection import SSRFProtectionError, validate_and_resolve_url\nfrom lfx.utils.ssrf_transport import create_ssrf_protected_client\n\n# Define fields for each mode\nMODE_FIELDS = {\n \"URL\": [\n \"url_input\",\n \"method\",\n ],\n \"cURL\": [\"curl_input\"],\n}\n\n# Fields that should always be visible\nDEFAULT_FIELDS = [\"mode\"]\n\n\nclass APIRequestComponent(Component):\n display_name = \"API Request\"\n description = \"Make HTTP requests using URL or cURL commands.\"\n documentation: str = \"https://docs.langflow.org/api-request\"\n icon = \"Globe\"\n name = \"APIRequest\"\n\n inputs = [\n MessageTextInput(\n name=\"url_input\",\n display_name=\"URL\",\n info=\"Enter the URL for the request.\",\n advanced=False,\n tool_mode=True,\n ),\n MultilineInput(\n name=\"curl_input\",\n display_name=\"cURL\",\n info=(\n \"Paste a curl command to populate the fields. \"\n \"This will fill in the dictionary fields for headers and body.\"\n ),\n real_time_refresh=True,\n tool_mode=True,\n advanced=True,\n show=False,\n ),\n DropdownInput(\n name=\"method\",\n display_name=\"Method\",\n options=[\"GET\", \"POST\", \"PATCH\", \"PUT\", \"DELETE\"],\n value=\"GET\",\n info=\"The HTTP method to use.\",\n real_time_refresh=True,\n ),\n TabInput(\n name=\"mode\",\n display_name=\"Mode\",\n options=[\"URL\", \"cURL\"],\n value=\"URL\",\n info=\"Enable cURL mode to populate fields from a cURL command.\",\n real_time_refresh=True,\n ),\n DataInput(\n name=\"query_params\",\n display_name=\"Query Parameters\",\n info=\"The query parameters to append to the URL.\",\n advanced=True,\n ),\n TableInput(\n name=\"body\",\n display_name=\"Body\",\n info=\"The body to send with the request as a dictionary (for POST, PATCH, PUT).\",\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Key\",\n \"type\": \"str\",\n \"description\": \"Parameter name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"description\": \"Parameter value\",\n },\n ],\n value=[],\n input_types=[\"Data\", \"JSON\"],\n advanced=True,\n real_time_refresh=True,\n ),\n TableInput(\n name=\"headers\",\n display_name=\"Headers\",\n info=\"The headers to send with the request\",\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Header\",\n \"type\": \"str\",\n \"description\": \"Header name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"type\": \"str\",\n \"description\": \"Header value\",\n },\n ],\n value=[{\"key\": \"User-Agent\", \"value\": \"Langflow/1.0\"}],\n advanced=True,\n input_types=[\"Data\", \"JSON\"],\n real_time_refresh=True,\n ),\n IntInput(\n name=\"timeout\",\n display_name=\"Timeout\",\n value=30,\n info=\"The timeout to use for the request.\",\n advanced=True,\n ),\n BoolInput(\n name=\"follow_redirects\",\n display_name=\"Follow Redirects\",\n value=False,\n info=(\n \"Whether to follow HTTP redirects. \"\n \"WARNING: Enabling redirects may allow SSRF bypass attacks where a public URL \"\n \"redirects to internal resources. Only enable if you trust the target server. \"\n \"See OWASP SSRF Prevention Cheat Sheet for details.\"\n ),\n advanced=True,\n ),\n BoolInput(\n name=\"save_to_file\",\n display_name=\"Save to File\",\n value=False,\n info=\"Save the API response to a temporary file\",\n advanced=True,\n ),\n BoolInput(\n name=\"include_httpx_metadata\",\n display_name=\"Include HTTPx Metadata\",\n value=False,\n info=(\n \"Include properties such as headers, status_code, response_headers, \"\n \"and redirection_history in the output.\"\n ),\n advanced=True,\n ),\n ]\n\n outputs = [\n Output(display_name=\"API Response\", name=\"data\", method=\"make_api_request\"),\n ]\n\n def _parse_json_value(self, value: Any) -> Any:\n \"\"\"Parse a value that might be a JSON string.\"\"\"\n if not isinstance(value, str):\n return value\n\n try:\n parsed = json.loads(value)\n except json.JSONDecodeError:\n return value\n else:\n return parsed\n\n def _process_body(self, body: Any) -> dict:\n \"\"\"Process the body input into a valid dictionary.\"\"\"\n if body is None:\n return {}\n if hasattr(body, \"data\"):\n body = body.data\n if isinstance(body, dict):\n return self._process_dict_body(body)\n if isinstance(body, str):\n return self._process_string_body(body)\n if isinstance(body, list):\n return self._process_list_body(body)\n return {}\n\n def _process_dict_body(self, body: dict) -> dict:\n \"\"\"Process dictionary body by parsing JSON values.\"\"\"\n return {k: self._parse_json_value(v) for k, v in body.items()}\n\n def _process_string_body(self, body: str) -> dict:\n \"\"\"Process string body by attempting JSON parse.\"\"\"\n try:\n return self._process_body(json.loads(body))\n except json.JSONDecodeError:\n return {\"data\": body}\n\n def _process_list_body(self, body: list) -> dict:\n \"\"\"Process list body by converting to key-value dictionary.\"\"\"\n processed_dict = {}\n try:\n for item in body:\n # Unwrap Data objects\n current_item = item\n if hasattr(item, \"data\"):\n unwrapped_data = item.data\n # If the unwrapped data is a dict but not key-value format, use it directly\n if isinstance(unwrapped_data, dict) and not self._is_valid_key_value_item(unwrapped_data):\n return unwrapped_data\n current_item = unwrapped_data\n if not self._is_valid_key_value_item(current_item):\n continue\n key = current_item[\"key\"]\n value = self._parse_json_value(current_item[\"value\"])\n processed_dict[key] = value\n except (KeyError, TypeError, ValueError) as e:\n self.log(f\"Failed to process body list: {e}\")\n return {}\n return processed_dict\n\n def _is_valid_key_value_item(self, item: Any) -> bool:\n \"\"\"Check if an item is a valid key-value dictionary.\"\"\"\n return isinstance(item, dict) and \"key\" in item and \"value\" in item\n\n def parse_curl(self, curl: str, build_config: dotdict) -> dotdict:\n \"\"\"Parse a cURL command and update build configuration.\"\"\"\n try:\n parsed = parse_context(curl)\n\n # Update basic configuration\n url = parsed.url\n # Normalize URL before setting it\n url = self._normalize_url(url)\n\n build_config[\"url_input\"][\"value\"] = url\n build_config[\"method\"][\"value\"] = parsed.method.upper()\n\n # Process headers\n headers_list = [{\"key\": k, \"value\": v} for k, v in parsed.headers.items()]\n build_config[\"headers\"][\"value\"] = headers_list\n\n # Process body data\n if not parsed.data:\n build_config[\"body\"][\"value\"] = []\n elif parsed.data:\n try:\n json_data = json.loads(parsed.data)\n if isinstance(json_data, dict):\n body_list = [\n {\"key\": k, \"value\": json.dumps(v) if isinstance(v, dict | list) else str(v)}\n for k, v in json_data.items()\n ]\n build_config[\"body\"][\"value\"] = body_list\n else:\n build_config[\"body\"][\"value\"] = [{\"key\": \"data\", \"value\": json.dumps(json_data)}]\n except json.JSONDecodeError:\n build_config[\"body\"][\"value\"] = [{\"key\": \"data\", \"value\": parsed.data}]\n\n except Exception as exc:\n msg = f\"Error parsing curl: {exc}\"\n self.log(msg)\n raise ValueError(msg) from exc\n\n return build_config\n\n def _normalize_url(self, url: str) -> str:\n \"\"\"Normalize URL by adding https:// if no protocol is specified.\"\"\"\n if not url or not isinstance(url, str):\n msg = \"URL cannot be empty\"\n raise ValueError(msg)\n\n url = url.strip()\n if url.startswith((\"http://\", \"https://\")):\n return url\n return f\"https://{url}\"\n\n async def make_request(\n self,\n client: httpx.AsyncClient,\n method: str,\n url: str,\n headers: dict | None = None,\n body: Any = None,\n timeout: int = 5,\n *,\n follow_redirects: bool = True,\n save_to_file: bool = False,\n include_httpx_metadata: bool = False,\n ) -> Data:\n method = method.upper()\n if method not in {\"GET\", \"POST\", \"PATCH\", \"PUT\", \"DELETE\"}:\n msg = f\"Unsupported method: {method}\"\n raise ValueError(msg)\n\n processed_body = self._process_body(body)\n redirection_history = []\n\n try:\n # Prepare request parameters\n request_params = {\n \"method\": method,\n \"url\": url,\n \"headers\": headers,\n \"timeout\": timeout,\n \"follow_redirects\": follow_redirects,\n }\n # Only include body for methods that support it (GET must not have a body per HTTP spec)\n if method in {\"POST\", \"PATCH\", \"PUT\", \"DELETE\"} and processed_body is not None:\n request_params[\"json\"] = processed_body\n response = await client.request(**request_params)\n\n redirection_history = [\n {\n \"url\": redirect.headers.get(\"Location\", str(redirect.url)),\n \"status_code\": redirect.status_code,\n }\n for redirect in response.history\n ]\n\n is_binary, file_path = await self._response_info(response, with_file_path=save_to_file)\n response_headers = self._headers_to_dict(response.headers)\n\n # Base metadata\n metadata = {\n \"source\": url,\n \"status_code\": response.status_code,\n \"response_headers\": response_headers,\n }\n\n if redirection_history:\n metadata[\"redirection_history\"] = redirection_history\n\n if save_to_file:\n mode = \"wb\" if is_binary else \"w\"\n encoding = response.encoding if mode == \"w\" else None\n if file_path:\n await aiofiles_os.makedirs(file_path.parent, exist_ok=True)\n if is_binary:\n async with aiofiles.open(file_path, \"wb\") as f:\n await f.write(response.content)\n await f.flush()\n else:\n async with aiofiles.open(file_path, \"w\", encoding=encoding) as f:\n await f.write(response.text)\n await f.flush()\n metadata[\"file_path\"] = str(file_path)\n\n if include_httpx_metadata:\n metadata.update({\"headers\": headers})\n return Data(data=metadata)\n\n # Handle response content\n if is_binary:\n result = response.content\n else:\n try:\n result = response.json()\n except json.JSONDecodeError:\n self.log(\"Failed to decode JSON response\")\n result = response.text.encode(\"utf-8\")\n\n metadata[\"result\"] = result\n\n if include_httpx_metadata:\n metadata.update({\"headers\": headers})\n\n return Data(data=metadata)\n except (httpx.HTTPError, httpx.RequestError, httpx.TimeoutException) as exc:\n self.log(f\"Error making request to {url}\")\n return Data(\n data={\n \"source\": url,\n \"headers\": headers,\n \"status_code\": 500,\n \"error\": str(exc),\n **({\"redirection_history\": redirection_history} if redirection_history else {}),\n },\n )\n\n def add_query_params(self, url: str, params: dict) -> str:\n \"\"\"Add query parameters to URL efficiently.\"\"\"\n if not params:\n return url\n url_parts = list(urlparse(url))\n query = dict(parse_qsl(url_parts[4]))\n query.update(params)\n url_parts[4] = urlencode(query)\n return urlunparse(url_parts)\n\n def _headers_to_dict(self, headers: httpx.Headers) -> dict[str, str]:\n \"\"\"Convert HTTP headers to a dictionary with lowercased keys.\"\"\"\n return {k.lower(): v for k, v in headers.items()}\n\n def _process_headers(self, headers: Any) -> dict:\n \"\"\"Process the headers input into a valid dictionary.\"\"\"\n if headers is None:\n return {}\n if isinstance(headers, dict):\n return headers\n if isinstance(headers, list):\n return {item[\"key\"]: item[\"value\"] for item in headers if self._is_valid_key_value_item(item)}\n return {}\n\n async def make_api_request(self) -> Data:\n \"\"\"Make HTTP request with SSRF protection and DNS pinning.\n\n This method implements comprehensive SSRF (Server-Side Request Forgery) protection\n using DNS pinning to prevent DNS rebinding attacks. The protection works by:\n 1. Validating the URL and resolving DNS during security check\n 2. Pinning the validated IP address\n 3. Forcing the HTTP client to use the pinned IP for the actual request\n 4. Ignoring any subsequent DNS changes (prevents rebinding attacks)\n\n Returns:\n Data: Response data from the HTTP request\n\n Raises:\n ValueError: If URL is invalid or blocked by SSRF protection\n \"\"\"\n # Extract request parameters\n method = self.method\n url = self.url_input.strip() if isinstance(self.url_input, str) else \"\"\n headers = self.headers or {}\n body = self.body or {}\n timeout = self.timeout\n follow_redirects = self.follow_redirects\n save_to_file = self.save_to_file\n include_httpx_metadata = self.include_httpx_metadata\n\n # Security warning: HTTP redirects can bypass SSRF protection\n # A public URL could redirect to an internal resource\n if follow_redirects:\n self.log(\n \"Security Warning: HTTP redirects are enabled. This may allow SSRF bypass attacks \"\n \"where a public URL redirects to internal resources (e.g., cloud metadata endpoints). \"\n \"Only enable this if you trust the target server.\"\n )\n\n # Normalize URL (add https:// if no protocol specified)\n url = self._normalize_url(url)\n\n # Basic URL format validation\n if not validators.url(url):\n msg = f\"Invalid URL provided: {url}\"\n raise ValueError(msg)\n\n # ============================================================================\n # SSRF Protection with DNS Pinning\n # ============================================================================\n # This prevents DNS rebinding attacks by:\n # 1. Resolving DNS and validating IPs during security check\n # 2. Pinning the validated IP address\n # 3. Using a custom HTTP transport that forces use of the pinned IP\n # 4. Ignoring any new DNS resolutions (prevents rebinding)\n #\n # Without DNS pinning, an attacker could:\n # - First DNS lookup: returns public IP (passes validation)\n # - Second DNS lookup: returns internal IP (bypasses protection)\n # - Attack succeeds: accesses internal services\n #\n # With DNS pinning:\n # - First DNS lookup: returns public IP (passes validation)\n # - IP is pinned: \"example.com = 93.184.216.34\"\n # - HTTP request: uses pinned IP directly (no new DNS lookup)\n # - Attack fails: even if DNS changes, we use the validated IP\n # ============================================================================\n\n try:\n # Validate URL and get validated IPs for DNS pinning\n _validated_url, validated_ips = validate_and_resolve_url(url)\n\n # Log DNS pinning information for security auditing\n if validated_ips:\n self.log(f\"SSRF Protection: Using DNS pinning with {len(validated_ips)} validated IP(s)\")\n\n except SSRFProtectionError as e:\n # SSRF protection blocked the request (private IP, internal network, etc.)\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n\n # Process query parameters (from string or Data object)\n if isinstance(self.query_params, str):\n query_params = dict(parse_qsl(self.query_params))\n else:\n query_params = self.query_params.data if self.query_params else {}\n\n # Process headers and body into proper format\n headers = self._process_headers(headers)\n body = self._process_body(body)\n url = self.add_query_params(url, query_params)\n\n # ============================================================================\n # Create HTTP Client with DNS Pinning (if SSRF protection enabled)\n # ============================================================================\n from lfx.utils.ssrf_protection import is_ssrf_protection_enabled\n\n if is_ssrf_protection_enabled() and validated_ips:\n # SSRF protection is enabled and DNS pinning is needed\n # Extract hostname from the final URL (after query params added)\n hostname = urlparse(url).hostname\n\n if hostname:\n # Create client with DNS pinning to prevent rebinding attacks\n # The custom transport will try validated IPs in order (supports dual-stack/load balancing)\n # while preserving the Host header for virtual hosting/SNI\n async with create_ssrf_protected_client(\n hostname=hostname,\n validated_ips=validated_ips, # Pass all validated IPs\n ) as client:\n result = await self.make_request(\n client,\n method,\n url,\n headers,\n body,\n timeout,\n follow_redirects=follow_redirects,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n else:\n # Hostname extraction failed - fallback to normal client\n # This should rarely happen as URL was already validated\n async with httpx.AsyncClient() as client:\n result = await self.make_request(\n client,\n method,\n url,\n headers,\n body,\n timeout,\n follow_redirects=follow_redirects,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n else:\n # No DNS pinning needed - use normal client\n # This happens when SSRF protection is disabled or host is allowlisted\n # - SSRF protection is disabled\n # - Host is in the allowlist (e.g., localhost for Ollama)\n # - Direct IP address was used (no DNS to pin)\n async with httpx.AsyncClient() as client:\n result = await self.make_request(\n client,\n method,\n url,\n headers,\n body,\n timeout,\n follow_redirects=follow_redirects,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n\n self.status = result\n return result\n\n def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:\n \"\"\"Update the build config based on the selected mode.\"\"\"\n if field_name != \"mode\":\n if field_name == \"curl_input\" and self.mode == \"cURL\" and self.curl_input:\n return self.parse_curl(self.curl_input, build_config)\n return build_config\n\n if field_value == \"cURL\":\n set_field_display(build_config, \"curl_input\", value=True)\n if build_config[\"curl_input\"][\"value\"]:\n try:\n build_config = self.parse_curl(build_config[\"curl_input\"][\"value\"], build_config)\n except ValueError as e:\n self.log(f\"Failed to parse cURL input: {e}\")\n else:\n set_field_display(build_config, \"curl_input\", value=False)\n\n return set_current_fields(\n build_config=build_config,\n action_fields=MODE_FIELDS,\n selected_action=field_value,\n default_fields=DEFAULT_FIELDS,\n func=set_field_advanced,\n default_value=True,\n )\n\n async def _response_info(\n self, response: httpx.Response, *, with_file_path: bool = False\n ) -> tuple[bool, Path | None]:\n \"\"\"Determine the file path and whether the response content is binary.\n\n Args:\n response (Response): The HTTP response object.\n with_file_path (bool): Whether to save the response content to a file.\n\n Returns:\n Tuple[bool, Path | None]:\n A tuple containing a boolean indicating if the content is binary and the full file path (if applicable).\n \"\"\"\n content_type = response.headers.get(\"Content-Type\", \"\")\n is_binary = \"application/octet-stream\" in content_type or \"application/binary\" in content_type\n\n if not with_file_path:\n return is_binary, None\n\n component_temp_dir = Path(tempfile.gettempdir()) / self.__class__.__name__\n\n # Create directory asynchronously\n await aiofiles_os.makedirs(component_temp_dir, exist_ok=True)\n\n filename = None\n if \"Content-Disposition\" in response.headers:\n content_disposition = response.headers[\"Content-Disposition\"]\n filename_match = re.search(r'filename=\"(.+?)\"', content_disposition)\n if filename_match:\n extracted_filename = filename_match.group(1)\n filename = extracted_filename\n\n # Step 3: Infer file extension or use part of the request URL if no filename\n if not filename:\n # Extract the last segment of the URL path\n url_path = urlparse(str(response.request.url) if response.request else \"\").path\n base_name = Path(url_path).name # Get the last segment of the path\n if not base_name: # If the path ends with a slash or is empty\n base_name = \"response\"\n\n # Infer file extension\n content_type_to_extension = {\n \"text/plain\": \".txt\",\n \"application/json\": \".json\",\n \"image/jpeg\": \".jpg\",\n \"image/png\": \".png\",\n \"application/octet-stream\": \".bin\",\n }\n extension = content_type_to_extension.get(content_type, \".bin\" if is_binary else \".txt\")\n filename = f\"{base_name}{extension}\"\n\n # Step 4: Define the full file path\n file_path = component_temp_dir / filename\n\n # Step 5: Check if file exists asynchronously and handle accordingly\n try:\n # Try to create the file exclusively (x mode) to check existence\n async with aiofiles.open(file_path, \"x\") as _:\n pass # File created successfully, we can use this path\n except FileExistsError:\n # If file exists, append a timestamp to the filename\n timestamp = datetime.now(timezone.utc).strftime(\"%Y%m%d%H%M%S%f\")\n file_path = component_temp_dir / f\"{timestamp}-{filename}\"\n\n return is_binary, file_path\n" + "value": "import json\nimport re\nimport tempfile\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom typing import Any\nfrom urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse\n\nimport aiofiles\nimport aiofiles.os as aiofiles_os\nimport httpx\nimport validators\n\nfrom lfx.base.curl.parse import parse_context\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import TabInput\nfrom lfx.io import (\n BoolInput,\n DataInput,\n DropdownInput,\n IntInput,\n MessageTextInput,\n MultilineInput,\n Output,\n TableInput,\n)\nfrom lfx.schema.data import Data\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.utils.component_utils import set_current_fields, set_field_advanced, set_field_display\n\n# SSRF Protection imports - for preventing Server-Side Request Forgery attacks\nfrom lfx.utils.ssrf_protection import (\n SSRFProtectionError,\n is_ssrf_protection_enabled,\n validate_and_resolve_url,\n)\nfrom lfx.utils.ssrf_transport import create_ssrf_protected_client\n\n# Define fields for each mode\nMODE_FIELDS = {\n \"URL\": [\n \"url_input\",\n \"method\",\n ],\n \"cURL\": [\"curl_input\"],\n}\n\n# Fields that should always be visible\nDEFAULT_FIELDS = [\"mode\"]\n\n# HTTP redirect status codes (RFC 9110).\nHTTP_MOVED_PERMANENTLY = 301\nHTTP_FOUND = 302\nHTTP_SEE_OTHER = 303\nHTTP_TEMPORARY_REDIRECT = 307\nHTTP_PERMANENT_REDIRECT = 308\n\n# Maximum number of redirects to follow when re-validating each hop (matches httpx's default).\nMAX_REDIRECTS = 20\n\n# HTTP status codes that represent a redirect carrying a Location header.\nREDIRECT_STATUS_CODES = frozenset(\n {\n HTTP_MOVED_PERMANENTLY,\n HTTP_FOUND,\n HTTP_SEE_OTHER,\n HTTP_TEMPORARY_REDIRECT,\n HTTP_PERMANENT_REDIRECT,\n }\n)\n\n\nclass APIRequestComponent(Component):\n display_name = \"API Request\"\n description = \"Make HTTP requests using URL or cURL commands.\"\n documentation: str = \"https://docs.langflow.org/api-request\"\n icon = \"Globe\"\n name = \"APIRequest\"\n\n inputs = [\n MessageTextInput(\n name=\"url_input\",\n display_name=\"URL\",\n info=\"Enter the URL for the request.\",\n advanced=False,\n tool_mode=True,\n ),\n MultilineInput(\n name=\"curl_input\",\n display_name=\"cURL\",\n info=(\n \"Paste a curl command to populate the fields. \"\n \"This will fill in the dictionary fields for headers and body.\"\n ),\n real_time_refresh=True,\n tool_mode=True,\n advanced=True,\n show=False,\n ),\n DropdownInput(\n name=\"method\",\n display_name=\"Method\",\n options=[\"GET\", \"POST\", \"PATCH\", \"PUT\", \"DELETE\"],\n value=\"GET\",\n info=\"The HTTP method to use.\",\n real_time_refresh=True,\n ),\n TabInput(\n name=\"mode\",\n display_name=\"Mode\",\n options=[\"URL\", \"cURL\"],\n value=\"URL\",\n info=\"Enable cURL mode to populate fields from a cURL command.\",\n real_time_refresh=True,\n ),\n DataInput(\n name=\"query_params\",\n display_name=\"Query Parameters\",\n info=\"The query parameters to append to the URL.\",\n advanced=True,\n ),\n TableInput(\n name=\"body\",\n display_name=\"Body\",\n info=\"The body to send with the request as a dictionary (for POST, PATCH, PUT).\",\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Key\",\n \"type\": \"str\",\n \"description\": \"Parameter name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"description\": \"Parameter value\",\n },\n ],\n value=[],\n input_types=[\"Data\", \"JSON\"],\n advanced=True,\n real_time_refresh=True,\n ),\n TableInput(\n name=\"headers\",\n display_name=\"Headers\",\n info=\"The headers to send with the request\",\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Header\",\n \"type\": \"str\",\n \"description\": \"Header name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"type\": \"str\",\n \"description\": \"Header value\",\n },\n ],\n value=[{\"key\": \"User-Agent\", \"value\": \"Langflow/1.0\"}],\n advanced=True,\n input_types=[\"Data\", \"JSON\"],\n real_time_refresh=True,\n ),\n IntInput(\n name=\"timeout\",\n display_name=\"Timeout\",\n value=30,\n info=\"The timeout to use for the request.\",\n advanced=True,\n ),\n BoolInput(\n name=\"follow_redirects\",\n display_name=\"Follow Redirects\",\n value=False,\n info=(\n \"Whether to follow HTTP redirects. \"\n \"WARNING: Enabling redirects may allow SSRF bypass attacks where a public URL \"\n \"redirects to internal resources. Only enable if you trust the target server. \"\n \"See OWASP SSRF Prevention Cheat Sheet for details.\"\n ),\n advanced=True,\n ),\n BoolInput(\n name=\"save_to_file\",\n display_name=\"Save to File\",\n value=False,\n info=\"Save the API response to a temporary file\",\n advanced=True,\n ),\n BoolInput(\n name=\"include_httpx_metadata\",\n display_name=\"Include HTTPx Metadata\",\n value=False,\n info=(\n \"Include properties such as headers, status_code, response_headers, \"\n \"and redirection_history in the output.\"\n ),\n advanced=True,\n ),\n ]\n\n outputs = [\n Output(display_name=\"API Response\", name=\"data\", method=\"make_api_request\"),\n ]\n\n def _parse_json_value(self, value: Any) -> Any:\n \"\"\"Parse a value that might be a JSON string.\"\"\"\n if not isinstance(value, str):\n return value\n\n try:\n parsed = json.loads(value)\n except json.JSONDecodeError:\n return value\n else:\n return parsed\n\n def _process_body(self, body: Any) -> dict:\n \"\"\"Process the body input into a valid dictionary.\"\"\"\n if body is None:\n return {}\n if hasattr(body, \"data\"):\n body = body.data\n if isinstance(body, dict):\n return self._process_dict_body(body)\n if isinstance(body, str):\n return self._process_string_body(body)\n if isinstance(body, list):\n return self._process_list_body(body)\n return {}\n\n def _process_dict_body(self, body: dict) -> dict:\n \"\"\"Process dictionary body by parsing JSON values.\"\"\"\n return {k: self._parse_json_value(v) for k, v in body.items()}\n\n def _process_string_body(self, body: str) -> dict:\n \"\"\"Process string body by attempting JSON parse.\"\"\"\n try:\n return self._process_body(json.loads(body))\n except json.JSONDecodeError:\n return {\"data\": body}\n\n def _process_list_body(self, body: list) -> dict:\n \"\"\"Process list body by converting to key-value dictionary.\"\"\"\n processed_dict = {}\n try:\n for item in body:\n # Unwrap Data objects\n current_item = item\n if hasattr(item, \"data\"):\n unwrapped_data = item.data\n # If the unwrapped data is a dict but not key-value format, use it directly\n if isinstance(unwrapped_data, dict) and not self._is_valid_key_value_item(unwrapped_data):\n return unwrapped_data\n current_item = unwrapped_data\n if not self._is_valid_key_value_item(current_item):\n continue\n key = current_item[\"key\"]\n value = self._parse_json_value(current_item[\"value\"])\n processed_dict[key] = value\n except (KeyError, TypeError, ValueError) as e:\n self.log(f\"Failed to process body list: {e}\")\n return {}\n return processed_dict\n\n def _is_valid_key_value_item(self, item: Any) -> bool:\n \"\"\"Check if an item is a valid key-value dictionary.\"\"\"\n return isinstance(item, dict) and \"key\" in item and \"value\" in item\n\n def parse_curl(self, curl: str, build_config: dotdict) -> dotdict:\n \"\"\"Parse a cURL command and update build configuration.\"\"\"\n try:\n parsed = parse_context(curl)\n\n # Update basic configuration\n url = parsed.url\n # Normalize URL before setting it\n url = self._normalize_url(url)\n\n build_config[\"url_input\"][\"value\"] = url\n build_config[\"method\"][\"value\"] = parsed.method.upper()\n\n # Process headers\n headers_list = [{\"key\": k, \"value\": v} for k, v in parsed.headers.items()]\n build_config[\"headers\"][\"value\"] = headers_list\n\n # Process body data\n if not parsed.data:\n build_config[\"body\"][\"value\"] = []\n elif parsed.data:\n try:\n json_data = json.loads(parsed.data)\n if isinstance(json_data, dict):\n body_list = [\n {\"key\": k, \"value\": json.dumps(v) if isinstance(v, dict | list) else str(v)}\n for k, v in json_data.items()\n ]\n build_config[\"body\"][\"value\"] = body_list\n else:\n build_config[\"body\"][\"value\"] = [{\"key\": \"data\", \"value\": json.dumps(json_data)}]\n except json.JSONDecodeError:\n build_config[\"body\"][\"value\"] = [{\"key\": \"data\", \"value\": parsed.data}]\n\n except Exception as exc:\n msg = f\"Error parsing curl: {exc}\"\n self.log(msg)\n raise ValueError(msg) from exc\n\n return build_config\n\n def _normalize_url(self, url: str) -> str:\n \"\"\"Normalize URL by adding https:// if no protocol is specified.\"\"\"\n if not url or not isinstance(url, str):\n msg = \"URL cannot be empty\"\n raise ValueError(msg)\n\n url = url.strip()\n if url.startswith((\"http://\", \"https://\")):\n return url\n return f\"https://{url}\"\n\n async def make_request(\n self,\n client: httpx.AsyncClient,\n method: str,\n url: str,\n headers: dict | None = None,\n body: Any = None,\n timeout: int = 5,\n *,\n follow_redirects: bool = False,\n save_to_file: bool = False,\n include_httpx_metadata: bool = False,\n ) -> Data:\n method = method.upper()\n if method not in {\"GET\", \"POST\", \"PATCH\", \"PUT\", \"DELETE\"}:\n msg = f\"Unsupported method: {method}\"\n raise ValueError(msg)\n\n processed_body = self._process_body(body)\n redirection_history = []\n\n try:\n # Prepare request parameters\n request_params = {\n \"method\": method,\n \"url\": url,\n \"headers\": headers,\n \"timeout\": timeout,\n \"follow_redirects\": follow_redirects,\n }\n # Only include body for methods that support it (GET must not have a body per HTTP spec)\n if method in {\"POST\", \"PATCH\", \"PUT\", \"DELETE\"} and processed_body is not None:\n request_params[\"json\"] = processed_body\n response = await client.request(**request_params)\n\n redirection_history = [\n {\n \"url\": redirect.headers.get(\"Location\", str(redirect.url)),\n \"status_code\": redirect.status_code,\n }\n for redirect in response.history\n ]\n\n return await self._build_response_data(\n response,\n url,\n headers,\n redirection_history,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n except (httpx.HTTPError, httpx.RequestError, httpx.TimeoutException) as exc:\n self.log(f\"Error making request to {url}\")\n return Data(\n data={\n \"source\": url,\n \"headers\": headers,\n \"status_code\": 500,\n \"error\": str(exc),\n **({\"redirection_history\": redirection_history} if redirection_history else {}),\n },\n )\n\n async def _build_response_data(\n self,\n response: httpx.Response,\n source_url: str,\n headers: dict | None,\n redirection_history: list,\n *,\n save_to_file: bool = False,\n include_httpx_metadata: bool = False,\n ) -> Data:\n \"\"\"Turn an httpx response into the component's ``Data`` output.\n\n Shared by the standard request path (``make_request``) and the redirect\n re-validation path (``_follow_redirects_with_validation``) so both produce\n identical metadata, optional file saving, and body decoding.\n \"\"\"\n is_binary, file_path = await self._response_info(response, with_file_path=save_to_file)\n response_headers = self._headers_to_dict(response.headers)\n\n # Base metadata\n metadata = {\n \"source\": source_url,\n \"status_code\": response.status_code,\n \"response_headers\": response_headers,\n }\n\n if redirection_history:\n metadata[\"redirection_history\"] = redirection_history\n\n if save_to_file:\n mode = \"wb\" if is_binary else \"w\"\n encoding = response.encoding if mode == \"w\" else None\n if file_path:\n await aiofiles_os.makedirs(file_path.parent, exist_ok=True)\n if is_binary:\n async with aiofiles.open(file_path, \"wb\") as f:\n await f.write(response.content)\n await f.flush()\n else:\n async with aiofiles.open(file_path, \"w\", encoding=encoding) as f:\n await f.write(response.text)\n await f.flush()\n metadata[\"file_path\"] = str(file_path)\n\n if include_httpx_metadata:\n metadata.update({\"headers\": headers})\n return Data(data=metadata)\n\n # Handle response content\n if is_binary:\n result = response.content\n else:\n try:\n result = response.json()\n except json.JSONDecodeError:\n self.log(\"Failed to decode JSON response\")\n result = response.text.encode(\"utf-8\")\n\n metadata[\"result\"] = result\n\n if include_httpx_metadata:\n metadata.update({\"headers\": headers})\n\n return Data(data=metadata)\n\n def add_query_params(self, url: str, params: dict) -> str:\n \"\"\"Add query parameters to URL efficiently.\"\"\"\n if not params:\n return url\n url_parts = list(urlparse(url))\n query = dict(parse_qsl(url_parts[4]))\n query.update(params)\n url_parts[4] = urlencode(query)\n return urlunparse(url_parts)\n\n def _headers_to_dict(self, headers: httpx.Headers) -> dict[str, str]:\n \"\"\"Convert HTTP headers to a dictionary with lowercased keys.\"\"\"\n return {k.lower(): v for k, v in headers.items()}\n\n def _process_headers(self, headers: Any) -> dict:\n \"\"\"Process the headers input into a valid dictionary.\"\"\"\n if headers is None:\n return {}\n if isinstance(headers, dict):\n return headers\n if isinstance(headers, list):\n return {item[\"key\"]: item[\"value\"] for item in headers if self._is_valid_key_value_item(item)}\n return {}\n\n async def make_api_request(self) -> Data:\n \"\"\"Make HTTP request with SSRF protection and DNS pinning.\n\n This method implements comprehensive SSRF (Server-Side Request Forgery) protection\n using DNS pinning to prevent DNS rebinding attacks. The protection works by:\n 1. Validating the URL and resolving DNS during security check\n 2. Pinning the validated IP address\n 3. Forcing the HTTP client to use the pinned IP for the actual request\n 4. Ignoring any subsequent DNS changes (prevents rebinding attacks)\n\n Returns:\n Data: Response data from the HTTP request\n\n Raises:\n ValueError: If URL is invalid or blocked by SSRF protection\n \"\"\"\n # Extract request parameters\n method = self.method\n url = self.url_input.strip() if isinstance(self.url_input, str) else \"\"\n headers = self.headers or {}\n body = self.body or {}\n timeout = self.timeout\n follow_redirects = self.follow_redirects\n save_to_file = self.save_to_file\n include_httpx_metadata = self.include_httpx_metadata\n\n # Security warning: HTTP redirects can bypass SSRF protection\n # A public URL could redirect to an internal resource\n if follow_redirects:\n self.log(\n \"Security Warning: HTTP redirects are enabled. This may allow SSRF bypass attacks \"\n \"where a public URL redirects to internal resources (e.g., cloud metadata endpoints). \"\n \"Only enable this if you trust the target server.\"\n )\n\n # Normalize URL (add https:// if no protocol specified)\n url = self._normalize_url(url)\n\n # Basic URL format validation\n if not validators.url(url):\n msg = f\"Invalid URL provided: {url}\"\n raise ValueError(msg)\n\n # ============================================================================\n # SSRF Protection with DNS Pinning\n # ============================================================================\n # This prevents DNS rebinding attacks by:\n # 1. Resolving DNS and validating IPs during security check\n # 2. Pinning the validated IP address\n # 3. Using a custom HTTP transport that forces use of the pinned IP\n # 4. Ignoring any new DNS resolutions (prevents rebinding)\n #\n # Without DNS pinning, an attacker could:\n # - First DNS lookup: returns public IP (passes validation)\n # - Second DNS lookup: returns internal IP (bypasses protection)\n # - Attack succeeds: accesses internal services\n #\n # With DNS pinning:\n # - First DNS lookup: returns public IP (passes validation)\n # - IP is pinned: \"example.com = 93.184.216.34\"\n # - HTTP request: uses pinned IP directly (no new DNS lookup)\n # - Attack fails: even if DNS changes, we use the validated IP\n # ============================================================================\n\n try:\n # Validate URL and get validated IPs for DNS pinning\n _validated_url, validated_ips = validate_and_resolve_url(url)\n\n # Log DNS pinning information for security auditing\n if validated_ips:\n self.log(f\"SSRF Protection: Using DNS pinning with {len(validated_ips)} validated IP(s)\")\n\n except SSRFProtectionError as e:\n # SSRF protection blocked the request (private IP, internal network, etc.)\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n\n # Process query parameters (from string or Data object)\n if isinstance(self.query_params, str):\n query_params = dict(parse_qsl(self.query_params))\n else:\n query_params = self.query_params.data if self.query_params else {}\n\n # Process headers and body into proper format\n headers = self._process_headers(headers)\n body = self._process_body(body)\n url = self.add_query_params(url, query_params)\n\n # ============================================================================\n # Execute the request (re-validating any redirects when SSRF protection is on)\n # ============================================================================\n # When SSRF protection is enabled we must NOT let httpx auto-follow redirects:\n # a validated public URL can redirect to an internal address (loopback, RFC1918,\n # link-local / cloud metadata) that was never checked, bypassing both the initial\n # validation and DNS pinning. Instead we follow redirects manually so every hop\n # is re-validated with the same denylist + DNS pinning. When protection is\n # disabled, we preserve the previous behavior and let httpx handle redirects.\n if is_ssrf_protection_enabled() and follow_redirects:\n result = await self._follow_redirects_with_validation(\n method,\n url,\n headers,\n body,\n timeout,\n validated_ips,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n else:\n # No redirect re-validation needed:\n # - SSRF protection is disabled (user opted out), or\n # - redirects are disabled, so httpx makes a single request.\n # DNS pinning still applies to the single request when protection is enabled\n # and the host resolved to validated IPs.\n async with self._build_http_client(url, validated_ips) as client:\n result = await self.make_request(\n client,\n method,\n url,\n headers,\n body,\n timeout,\n follow_redirects=follow_redirects,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n\n self.status = result\n return result\n\n def _build_http_client(self, url: str, validated_ips: list[str]) -> httpx.AsyncClient:\n \"\"\"Create an HTTP client, pinning DNS to validated IPs when SSRF protection applies.\n\n Args:\n url: The request URL whose hostname will be pinned.\n validated_ips: IPs validated by ``validate_and_resolve_url`` for this hop.\n\n Returns:\n httpx.AsyncClient: A client that pins DNS to ``validated_ips`` (preventing\n rebinding) when SSRF protection is enabled and the hop has validated IPs;\n otherwise a standard client (protection disabled, allowlisted host, or\n hostname extraction failure).\n \"\"\"\n if is_ssrf_protection_enabled() and validated_ips:\n # Extract hostname from the URL so the custom transport can pin it while\n # preserving the Host header for virtual hosting / TLS SNI.\n hostname = urlparse(url).hostname\n if hostname:\n # The custom transport tries validated IPs in order (dual-stack / LB).\n return create_ssrf_protected_client(hostname=hostname, validated_ips=validated_ips)\n return httpx.AsyncClient()\n\n @staticmethod\n def _method_for_redirect(method: str, status_code: int) -> str:\n \"\"\"Return the HTTP method to use after a redirect, mirroring httpx semantics.\n\n A 303 (See Other) always becomes GET; 301/302 downgrade POST to GET for\n browser compatibility; 307/308 preserve the original method (and body).\n \"\"\"\n method = method.upper()\n if status_code == HTTP_SEE_OTHER and method != \"HEAD\":\n return \"GET\"\n if status_code in (HTTP_MOVED_PERMANENTLY, HTTP_FOUND) and method == \"POST\":\n return \"GET\"\n return method\n\n @staticmethod\n def _headers_for_redirect(headers: dict | None, current_url: str, next_url: str) -> dict | None:\n \"\"\"Drop sensitive headers when a redirect crosses to a different host.\n\n Mirrors httpx's auto-follow behavior so manually following redirects does not\n leak credentials (Authorization / Cookie) to a different host than the one the\n caller intended them for. Same-host redirects keep all headers.\n \"\"\"\n if not headers:\n return headers\n if urlparse(current_url).hostname == urlparse(next_url).hostname:\n return headers\n sensitive = {\"authorization\", \"proxy-authorization\", \"cookie\"}\n return {k: v for k, v in headers.items() if k.lower() not in sensitive}\n\n async def _follow_redirects_with_validation(\n self,\n method: str,\n url: str,\n headers: dict | None,\n body: Any,\n timeout: int,\n validated_ips: list[str],\n *,\n save_to_file: bool = False,\n include_httpx_metadata: bool = False,\n ) -> Data:\n \"\"\"Make the request and follow redirects manually, re-validating every hop.\n\n This closes an SSRF bypass: with ``follow_redirects`` enabled, httpx would\n otherwise auto-follow a redirect from a validated public URL to an internal\n address that was never checked. Here each redirect ``Location`` is resolved\n (relative locations included) and re-validated with ``validate_and_resolve_url``\n — the same private/loopback/link-local denylist and DNS pinning applied to the\n initial request — before any connection to it is made. A blocked hop raises\n ``ValueError``; the number of redirects is capped at ``MAX_REDIRECTS``.\n \"\"\"\n method = method.upper()\n if method not in {\"GET\", \"POST\", \"PATCH\", \"PUT\", \"DELETE\"}:\n msg = f\"Unsupported method: {method}\"\n raise ValueError(msg)\n\n processed_body = self._process_body(body)\n current_url = url\n current_ips = validated_ips\n redirection_history: list[dict] = []\n\n for _ in range(MAX_REDIRECTS + 1):\n request_params: dict[str, Any] = {\n \"method\": method,\n \"url\": current_url,\n \"headers\": headers,\n \"timeout\": timeout,\n # Never let httpx follow redirects itself; each hop is validated below.\n \"follow_redirects\": False,\n }\n # Only include body for methods that support it (GET must not have a body).\n if method in {\"POST\", \"PATCH\", \"PUT\", \"DELETE\"} and processed_body is not None:\n request_params[\"json\"] = processed_body\n\n try:\n async with self._build_http_client(current_url, current_ips) as client:\n response = await client.request(**request_params)\n except (httpx.HTTPError, httpx.RequestError, httpx.TimeoutException) as exc:\n self.log(f\"Error making request to {current_url}\")\n return Data(\n data={\n \"source\": url,\n \"headers\": headers,\n \"status_code\": 500,\n \"error\": str(exc),\n **({\"redirection_history\": redirection_history} if redirection_history else {}),\n },\n )\n\n location = response.headers.get(\"Location\")\n if response.status_code in REDIRECT_STATUS_CODES and location:\n # Resolve relative redirects against the current URL.\n next_url = urljoin(current_url, location)\n redirection_history.append({\"url\": location, \"status_code\": response.status_code})\n\n # Re-validate the redirect target with the same SSRF denylist + DNS pinning.\n # Non-http(s) schemes, private/loopback/link-local hosts, and hostnames that\n # resolve to blocked IPs all raise SSRFProtectionError here.\n try:\n _validated_url, current_ips = validate_and_resolve_url(next_url)\n except SSRFProtectionError as e:\n msg = f\"SSRF Protection: blocked redirect to {next_url}: {e}\"\n raise ValueError(msg) from e\n\n method = self._method_for_redirect(method, response.status_code)\n headers = self._headers_for_redirect(headers, current_url, next_url)\n current_url = next_url\n continue\n\n # Not a redirect (or no Location header) - this is the final response.\n return await self._build_response_data(\n response,\n url,\n headers,\n redirection_history,\n save_to_file=save_to_file,\n include_httpx_metadata=include_httpx_metadata,\n )\n\n msg = f\"SSRF Protection: exceeded the maximum of {MAX_REDIRECTS} redirects while requesting {url}\"\n raise ValueError(msg)\n\n def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:\n \"\"\"Update the build config based on the selected mode.\"\"\"\n if field_name != \"mode\":\n if field_name == \"curl_input\" and self.mode == \"cURL\" and self.curl_input:\n return self.parse_curl(self.curl_input, build_config)\n return build_config\n\n if field_value == \"cURL\":\n set_field_display(build_config, \"curl_input\", value=True)\n if build_config[\"curl_input\"][\"value\"]:\n try:\n build_config = self.parse_curl(build_config[\"curl_input\"][\"value\"], build_config)\n except ValueError as e:\n self.log(f\"Failed to parse cURL input: {e}\")\n else:\n set_field_display(build_config, \"curl_input\", value=False)\n\n return set_current_fields(\n build_config=build_config,\n action_fields=MODE_FIELDS,\n selected_action=field_value,\n default_fields=DEFAULT_FIELDS,\n func=set_field_advanced,\n default_value=True,\n )\n\n async def _response_info(\n self, response: httpx.Response, *, with_file_path: bool = False\n ) -> tuple[bool, Path | None]:\n \"\"\"Determine the file path and whether the response content is binary.\n\n Args:\n response (Response): The HTTP response object.\n with_file_path (bool): Whether to save the response content to a file.\n\n Returns:\n Tuple[bool, Path | None]:\n A tuple containing a boolean indicating if the content is binary and the full file path (if applicable).\n \"\"\"\n content_type = response.headers.get(\"Content-Type\", \"\")\n is_binary = \"application/octet-stream\" in content_type or \"application/binary\" in content_type\n\n if not with_file_path:\n return is_binary, None\n\n component_temp_dir = Path(tempfile.gettempdir()) / self.__class__.__name__\n\n # Create directory asynchronously\n await aiofiles_os.makedirs(component_temp_dir, exist_ok=True)\n\n filename = None\n if \"Content-Disposition\" in response.headers:\n content_disposition = response.headers[\"Content-Disposition\"]\n filename_match = re.search(r'filename=\"(.+?)\"', content_disposition)\n if filename_match:\n extracted_filename = filename_match.group(1)\n filename = extracted_filename\n\n # Step 3: Infer file extension or use part of the request URL if no filename\n if not filename:\n # Extract the last segment of the URL path\n url_path = urlparse(str(response.request.url) if response.request else \"\").path\n base_name = Path(url_path).name # Get the last segment of the path\n if not base_name: # If the path ends with a slash or is empty\n base_name = \"response\"\n\n # Infer file extension\n content_type_to_extension = {\n \"text/plain\": \".txt\",\n \"application/json\": \".json\",\n \"image/jpeg\": \".jpg\",\n \"image/png\": \".png\",\n \"application/octet-stream\": \".bin\",\n }\n extension = content_type_to_extension.get(content_type, \".bin\" if is_binary else \".txt\")\n filename = f\"{base_name}{extension}\"\n\n # Step 4: Define the full file path\n file_path = component_temp_dir / filename\n\n # Step 5: Check if file exists asynchronously and handle accordingly\n try:\n # Try to create the file exclusively (x mode) to check existence\n async with aiofiles.open(file_path, \"x\") as _:\n pass # File created successfully, we can use this path\n except FileExistsError:\n # If file exists, append a timestamp to the filename\n timestamp = datetime.now(timezone.utc).strftime(\"%Y%m%d%H%M%S%f\")\n file_path = component_temp_dir / f\"{timestamp}-{filename}\"\n\n return is_binary, file_path\n" }, "curl_input": { "_input_type": "MultilineInput", diff --git a/src/backend/tests/unit/components/data_source/test_api_request_component.py b/src/backend/tests/unit/components/data_source/test_api_request_component.py index 37d21819c63f..0f88a484ec24 100644 --- a/src/backend/tests/unit/components/data_source/test_api_request_component.py +++ b/src/backend/tests/unit/components/data_source/test_api_request_component.py @@ -1,4 +1,6 @@ +import ipaddress import os +import socket from pathlib import Path from unittest.mock import patch @@ -684,3 +686,209 @@ async def test_invalid_url_raises_error(self, component): async def test_follow_redirects_disabled_by_default(self, component): """Test that follow_redirects is disabled by default for security.""" assert component.follow_redirects is False + + +def _resolve_public(host, *_args, **_kwargs): + """socket.getaddrinfo stub: hostnames resolve to a public IP, literal IPs to themselves. + + Mirrors real DNS: the public redirector hostnames map to a public address, while a + literal IP (e.g. an internal 127.0.0.1 / 192.168.x redirect target) resolves to + itself so SSRF validation still classifies it as internal. + """ + try: + ipaddress.ip_address(host) + except ValueError: + ip = "93.184.216.34" # hostname -> public IP + else: + ip = host # literal IP -> itself + family = socket.AF_INET6 if ":" in ip else socket.AF_INET + return [(family, socket.SOCK_STREAM, 6, "", (ip, 0))] + + +class TestAPIRequestRedirectSSRFProtection: + """Regression tests for the SSRF redirect-following bypass. + + When SSRF protection is enabled, a validated public URL must not be able to reach + internal services by redirecting to them. The component follows redirects manually + and re-validates every hop with the same denylist + DNS pinning used for the + initial request, instead of trusting httpx to auto-follow unvalidated redirects. + """ + + @pytest.fixture + def component(self): + """Return a component configured to follow redirects.""" + return APIRequestComponent( + url_input="http://public.example.com/start", + method="GET", + headers=[], + body=[], + timeout=30, + follow_redirects=True, + save_to_file=False, + include_httpx_metadata=True, + mode="URL", + curl_input="", + query_params={}, + ) + + @respx.mock + @pytest.mark.parametrize( + ("internal_url", "description"), + [ + ("http://127.0.0.1:9999/secret", "loopback"), + ("http://192.168.0.10/admin", "rfc1918-192"), + ("http://10.0.0.5/internal", "rfc1918-10"), + ("http://172.16.0.9/internal", "rfc1918-172"), + ("http://169.254.169.254/latest/meta-data/", "link-local-metadata"), + ("http://0.0.0.0:8080/admin", "unspecified"), + ], + ) + async def test_redirect_to_internal_address_is_blocked(self, component, internal_url, description): + """A public URL that redirects to an internal address must be blocked, not followed.""" + marker = "INTERNAL_REDIRECT_SECRET_7a51f4" + respx.get("http://public.example.com/start").mock( + return_value=Response(302, headers={"Location": internal_url}) + ) + # If the fix regresses, the component would follow the redirect and serve this marker. + internal_route = respx.get(internal_url).mock(return_value=Response(200, text=marker)) + + with ( + patch("socket.getaddrinfo", side_effect=_resolve_public), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="blocked redirect"), + ): + await component.make_api_request() + + assert not internal_route.called, f"Redirect to {description} ({internal_url}) must not be followed" + + @respx.mock + async def test_redirect_scheme_change_is_blocked(self, component): + """A redirect that switches to a non-http(s) scheme (e.g. file://) must be blocked.""" + respx.get("http://public.example.com/start").mock( + return_value=Response(302, headers={"Location": "file:///etc/passwd"}) + ) + + with ( + patch("socket.getaddrinfo", side_effect=_resolve_public), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="blocked redirect"), + ): + await component.make_api_request() + + @respx.mock + async def test_redirect_to_hostname_resolving_internal_is_blocked(self, component): + """A redirect to a hostname that resolves to an internal IP must be blocked. + + Covers the DNS-rebinding-across-hops vector at the validation layer: the redirect + target host resolves to a blocked address and is rejected before any connection. + """ + + def resolve(host, *_args, **_kwargs): + ip = "127.0.0.1" if host == "internal.example.com" else "93.184.216.34" + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 0))] + + respx.get("http://public.example.com/start").mock( + return_value=Response(302, headers={"Location": "http://internal.example.com/secret"}) + ) + internal_route = respx.get("http://internal.example.com/secret").mock(return_value=Response(200, text="SECRET")) + + with ( + patch("socket.getaddrinfo", side_effect=resolve), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="blocked redirect"), + ): + await component.make_api_request() + + assert not internal_route.called + + @respx.mock + async def test_chained_public_redirects_are_followed(self, component): + """Legitimate public-to-public redirect chains still work (redirects are not disabled).""" + component.url_input = "http://hop1.example.com/a" + respx.get("http://hop1.example.com/a").mock( + return_value=Response(302, headers={"Location": "http://hop2.example.com/b"}) + ) + respx.get("http://hop2.example.com/b").mock( + return_value=Response(307, headers={"Location": "http://hop3.example.com/c"}) + ) + respx.get("http://hop3.example.com/c").mock(return_value=Response(200, json={"status": "ok"})) + + with ( + patch("socket.getaddrinfo", side_effect=_resolve_public), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + ): + result = await component.make_api_request() + + assert isinstance(result, Data) + assert result.data["status_code"] == 200 + assert result.data["result"]["status"] == "ok" + assert result.data["redirection_history"] == [ + {"url": "http://hop2.example.com/b", "status_code": 302}, + {"url": "http://hop3.example.com/c", "status_code": 307}, + ] + + @respx.mock + async def test_too_many_redirects_raises(self, component): + """A redirect loop is bounded and raises instead of looping forever.""" + component.url_input = "http://loop.example.com/a" + respx.get("http://loop.example.com/a").mock( + return_value=Response(302, headers={"Location": "http://loop.example.com/b"}) + ) + respx.get("http://loop.example.com/b").mock( + return_value=Response(302, headers={"Location": "http://loop.example.com/a"}) + ) + + with ( + patch("socket.getaddrinfo", side_effect=_resolve_public), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="exceeded the maximum"), + ): + await component.make_api_request() + + @respx.mock + async def test_redirect_to_internal_allowed_when_protection_disabled(self, component): + """With SSRF protection disabled, redirect behavior is unchanged (user opted out).""" + respx.get("http://public.example.com/start").mock( + return_value=Response(302, headers={"Location": "http://127.0.0.1:9999/ok"}) + ) + respx.get("http://127.0.0.1:9999/ok").mock(return_value=Response(200, json={"status": "reached"})) + + with patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}): + result = await component.make_api_request() + + assert result.data["status_code"] == 200 + assert result.data["result"]["status"] == "reached" + + @respx.mock + async def test_sensitive_headers_dropped_on_cross_host_redirect(self, component): + """Authorization/Cookie must not be forwarded to a different host on redirect.""" + component.headers = [ + {"key": "Authorization", "value": "Bearer secret-token"}, + {"key": "X-Custom", "value": "keep-me"}, + ] + + respx.get("http://public.example.com/start").mock( + return_value=Response(302, headers={"Location": "http://other.example.com/next"}) + ) + final_route = respx.get("http://other.example.com/next").mock(return_value=Response(200, json={"ok": True})) + + with ( + patch("socket.getaddrinfo", side_effect=_resolve_public), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + ): + result = await component.make_api_request() + + assert isinstance(result, Data) + assert final_route.called + forwarded = final_route.calls.last.request.headers + assert "authorization" not in {k.lower() for k in forwarded}, "Authorization must be stripped cross-host" + assert forwarded.get("X-Custom") == "keep-me", "Non-sensitive headers should be preserved" + + def test_method_for_redirect_semantics(self): + """301/302/303 downgrade POST to GET; 307/308 preserve the method.""" + assert APIRequestComponent._method_for_redirect("POST", 301) == "GET" + assert APIRequestComponent._method_for_redirect("POST", 302) == "GET" + assert APIRequestComponent._method_for_redirect("POST", 303) == "GET" + assert APIRequestComponent._method_for_redirect("POST", 307) == "POST" + assert APIRequestComponent._method_for_redirect("POST", 308) == "POST" + assert APIRequestComponent._method_for_redirect("GET", 302) == "GET" diff --git a/src/backend/tests/unit/components/data_source/test_dns_rebinding.py b/src/backend/tests/unit/components/data_source/test_dns_rebinding.py index 5471b685f16d..ac25f635f724 100644 --- a/src/backend/tests/unit/components/data_source/test_dns_rebinding.py +++ b/src/backend/tests/unit/components/data_source/test_dns_rebinding.py @@ -372,3 +372,57 @@ async def mock_connect_tcp(self, host, port, **kwargs): # Verify the result (should succeed with second IP) assert isinstance(result, Data) assert result.data is not None + + @pytest.mark.asyncio + async def test_dns_pinning_applies_to_redirect_hops(self, component): + """Test that DNS pinning is enforced on every redirect hop, not just the first. + + Simulates a redirect from a validated public host to a second host. With + per-hop validation + pinning, both connections must go to the validated public + IP - a rebinding attacker who flips the redirect target to 127.0.0.1 after + validation can never cause a connection to the internal address. + """ + component.url_input = "http://public.example.com/start" + component.follow_redirects = True + + resolved = [] + + def mock_getaddrinfo(host, *_args, **_kwargs): + """Every host resolves to a public IP at validation time.""" + resolved.append(host) + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 0))] + + connect_calls = [] + redirect_response = [ + b"HTTP/1.1 302 Found\r\n", + b"Location: http://rebind.example.com/secret\r\n", + b"Content-Length: 0\r\n", + b"\r\n", + ] + final_response = [ + b"HTTP/1.1 200 OK\r\n", + b"Content-Type: application/json\r\n", + b"Content-Length: 15\r\n", + b"\r\n", + b'{"status":"ok"}', + ] + + async def mock_connect_tcp(self, host, port, **kwargs): + """Capture every IP connected to; first hop redirects, second succeeds.""" + connect_calls.append(host) + stream = redirect_response if len(connect_calls) == 1 else final_response + return httpcore.AsyncMockStream(list(stream)) + + with ( + patch("socket.getaddrinfo", side_effect=mock_getaddrinfo), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + patch.object(httpcore.AnyIOBackend, "connect_tcp", mock_connect_tcp), + ): + result = await component.make_api_request() + + assert isinstance(result, Data) + # Both the initial request and the redirect hop connected to the pinned public IP. + assert connect_calls == ["8.8.8.8", "8.8.8.8"], connect_calls + # The redirect target was resolved exactly once (validation only); httpx never + # re-resolved it, so a rebinding flip after validation has no effect. + assert resolved.count("rebind.example.com") == 1, resolved diff --git a/src/lfx/src/lfx/components/data_source/api_request.py b/src/lfx/src/lfx/components/data_source/api_request.py index 85350dc581c7..ba6dab8382e1 100644 --- a/src/lfx/src/lfx/components/data_source/api_request.py +++ b/src/lfx/src/lfx/components/data_source/api_request.py @@ -4,7 +4,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any -from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse +from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse import aiofiles import aiofiles.os as aiofiles_os @@ -29,7 +29,11 @@ from lfx.utils.component_utils import set_current_fields, set_field_advanced, set_field_display # SSRF Protection imports - for preventing Server-Side Request Forgery attacks -from lfx.utils.ssrf_protection import SSRFProtectionError, validate_and_resolve_url +from lfx.utils.ssrf_protection import ( + SSRFProtectionError, + is_ssrf_protection_enabled, + validate_and_resolve_url, +) from lfx.utils.ssrf_transport import create_ssrf_protected_client # Define fields for each mode @@ -44,6 +48,27 @@ # Fields that should always be visible DEFAULT_FIELDS = ["mode"] +# HTTP redirect status codes (RFC 9110). +HTTP_MOVED_PERMANENTLY = 301 +HTTP_FOUND = 302 +HTTP_SEE_OTHER = 303 +HTTP_TEMPORARY_REDIRECT = 307 +HTTP_PERMANENT_REDIRECT = 308 + +# Maximum number of redirects to follow when re-validating each hop (matches httpx's default). +MAX_REDIRECTS = 20 + +# HTTP status codes that represent a redirect carrying a Location header. +REDIRECT_STATUS_CODES = frozenset( + { + HTTP_MOVED_PERMANENTLY, + HTTP_FOUND, + HTTP_SEE_OTHER, + HTTP_TEMPORARY_REDIRECT, + HTTP_PERMANENT_REDIRECT, + } +) + class APIRequestComponent(Component): display_name = "API Request" @@ -306,7 +331,7 @@ async def make_request( body: Any = None, timeout: int = 5, *, - follow_redirects: bool = True, + follow_redirects: bool = False, save_to_file: bool = False, include_httpx_metadata: bool = False, ) -> Data: @@ -340,54 +365,14 @@ async def make_request( for redirect in response.history ] - is_binary, file_path = await self._response_info(response, with_file_path=save_to_file) - response_headers = self._headers_to_dict(response.headers) - - # Base metadata - metadata = { - "source": url, - "status_code": response.status_code, - "response_headers": response_headers, - } - - if redirection_history: - metadata["redirection_history"] = redirection_history - - if save_to_file: - mode = "wb" if is_binary else "w" - encoding = response.encoding if mode == "w" else None - if file_path: - await aiofiles_os.makedirs(file_path.parent, exist_ok=True) - if is_binary: - async with aiofiles.open(file_path, "wb") as f: - await f.write(response.content) - await f.flush() - else: - async with aiofiles.open(file_path, "w", encoding=encoding) as f: - await f.write(response.text) - await f.flush() - metadata["file_path"] = str(file_path) - - if include_httpx_metadata: - metadata.update({"headers": headers}) - return Data(data=metadata) - - # Handle response content - if is_binary: - result = response.content - else: - try: - result = response.json() - except json.JSONDecodeError: - self.log("Failed to decode JSON response") - result = response.text.encode("utf-8") - - metadata["result"] = result - - if include_httpx_metadata: - metadata.update({"headers": headers}) - - return Data(data=metadata) + return await self._build_response_data( + response, + url, + headers, + redirection_history, + save_to_file=save_to_file, + include_httpx_metadata=include_httpx_metadata, + ) except (httpx.HTTPError, httpx.RequestError, httpx.TimeoutException) as exc: self.log(f"Error making request to {url}") return Data( @@ -400,6 +385,71 @@ async def make_request( }, ) + async def _build_response_data( + self, + response: httpx.Response, + source_url: str, + headers: dict | None, + redirection_history: list, + *, + save_to_file: bool = False, + include_httpx_metadata: bool = False, + ) -> Data: + """Turn an httpx response into the component's ``Data`` output. + + Shared by the standard request path (``make_request``) and the redirect + re-validation path (``_follow_redirects_with_validation``) so both produce + identical metadata, optional file saving, and body decoding. + """ + is_binary, file_path = await self._response_info(response, with_file_path=save_to_file) + response_headers = self._headers_to_dict(response.headers) + + # Base metadata + metadata = { + "source": source_url, + "status_code": response.status_code, + "response_headers": response_headers, + } + + if redirection_history: + metadata["redirection_history"] = redirection_history + + if save_to_file: + mode = "wb" if is_binary else "w" + encoding = response.encoding if mode == "w" else None + if file_path: + await aiofiles_os.makedirs(file_path.parent, exist_ok=True) + if is_binary: + async with aiofiles.open(file_path, "wb") as f: + await f.write(response.content) + await f.flush() + else: + async with aiofiles.open(file_path, "w", encoding=encoding) as f: + await f.write(response.text) + await f.flush() + metadata["file_path"] = str(file_path) + + if include_httpx_metadata: + metadata.update({"headers": headers}) + return Data(data=metadata) + + # Handle response content + if is_binary: + result = response.content + else: + try: + result = response.json() + except json.JSONDecodeError: + self.log("Failed to decode JSON response") + result = response.text.encode("utf-8") + + metadata["result"] = result + + if include_httpx_metadata: + metadata.update({"headers": headers}) + + return Data(data=metadata) + def add_query_params(self, url: str, params: dict) -> str: """Add query parameters to URL efficiently.""" if not params: @@ -513,56 +563,32 @@ async def make_api_request(self) -> Data: url = self.add_query_params(url, query_params) # ============================================================================ - # Create HTTP Client with DNS Pinning (if SSRF protection enabled) + # Execute the request (re-validating any redirects when SSRF protection is on) # ============================================================================ - from lfx.utils.ssrf_protection import is_ssrf_protection_enabled - - if is_ssrf_protection_enabled() and validated_ips: - # SSRF protection is enabled and DNS pinning is needed - # Extract hostname from the final URL (after query params added) - hostname = urlparse(url).hostname - - if hostname: - # Create client with DNS pinning to prevent rebinding attacks - # The custom transport will try validated IPs in order (supports dual-stack/load balancing) - # while preserving the Host header for virtual hosting/SNI - async with create_ssrf_protected_client( - hostname=hostname, - validated_ips=validated_ips, # Pass all validated IPs - ) as client: - result = await self.make_request( - client, - method, - url, - headers, - body, - timeout, - follow_redirects=follow_redirects, - save_to_file=save_to_file, - include_httpx_metadata=include_httpx_metadata, - ) - else: - # Hostname extraction failed - fallback to normal client - # This should rarely happen as URL was already validated - async with httpx.AsyncClient() as client: - result = await self.make_request( - client, - method, - url, - headers, - body, - timeout, - follow_redirects=follow_redirects, - save_to_file=save_to_file, - include_httpx_metadata=include_httpx_metadata, - ) + # When SSRF protection is enabled we must NOT let httpx auto-follow redirects: + # a validated public URL can redirect to an internal address (loopback, RFC1918, + # link-local / cloud metadata) that was never checked, bypassing both the initial + # validation and DNS pinning. Instead we follow redirects manually so every hop + # is re-validated with the same denylist + DNS pinning. When protection is + # disabled, we preserve the previous behavior and let httpx handle redirects. + if is_ssrf_protection_enabled() and follow_redirects: + result = await self._follow_redirects_with_validation( + method, + url, + headers, + body, + timeout, + validated_ips, + save_to_file=save_to_file, + include_httpx_metadata=include_httpx_metadata, + ) else: - # No DNS pinning needed - use normal client - # This happens when SSRF protection is disabled or host is allowlisted - # - SSRF protection is disabled - # - Host is in the allowlist (e.g., localhost for Ollama) - # - Direct IP address was used (no DNS to pin) - async with httpx.AsyncClient() as client: + # No redirect re-validation needed: + # - SSRF protection is disabled (user opted out), or + # - redirects are disabled, so httpx makes a single request. + # DNS pinning still applies to the single request when protection is enabled + # and the host resolved to validated IPs. + async with self._build_http_client(url, validated_ips) as client: result = await self.make_request( client, method, @@ -578,6 +604,150 @@ async def make_api_request(self) -> Data: self.status = result return result + def _build_http_client(self, url: str, validated_ips: list[str]) -> httpx.AsyncClient: + """Create an HTTP client, pinning DNS to validated IPs when SSRF protection applies. + + Args: + url: The request URL whose hostname will be pinned. + validated_ips: IPs validated by ``validate_and_resolve_url`` for this hop. + + Returns: + httpx.AsyncClient: A client that pins DNS to ``validated_ips`` (preventing + rebinding) when SSRF protection is enabled and the hop has validated IPs; + otherwise a standard client (protection disabled, allowlisted host, or + hostname extraction failure). + """ + if is_ssrf_protection_enabled() and validated_ips: + # Extract hostname from the URL so the custom transport can pin it while + # preserving the Host header for virtual hosting / TLS SNI. + hostname = urlparse(url).hostname + if hostname: + # The custom transport tries validated IPs in order (dual-stack / LB). + return create_ssrf_protected_client(hostname=hostname, validated_ips=validated_ips) + return httpx.AsyncClient() + + @staticmethod + def _method_for_redirect(method: str, status_code: int) -> str: + """Return the HTTP method to use after a redirect, mirroring httpx semantics. + + A 303 (See Other) always becomes GET; 301/302 downgrade POST to GET for + browser compatibility; 307/308 preserve the original method (and body). + """ + method = method.upper() + if status_code == HTTP_SEE_OTHER and method != "HEAD": + return "GET" + if status_code in (HTTP_MOVED_PERMANENTLY, HTTP_FOUND) and method == "POST": + return "GET" + return method + + @staticmethod + def _headers_for_redirect(headers: dict | None, current_url: str, next_url: str) -> dict | None: + """Drop sensitive headers when a redirect crosses to a different host. + + Mirrors httpx's auto-follow behavior so manually following redirects does not + leak credentials (Authorization / Cookie) to a different host than the one the + caller intended them for. Same-host redirects keep all headers. + """ + if not headers: + return headers + if urlparse(current_url).hostname == urlparse(next_url).hostname: + return headers + sensitive = {"authorization", "proxy-authorization", "cookie"} + return {k: v for k, v in headers.items() if k.lower() not in sensitive} + + async def _follow_redirects_with_validation( + self, + method: str, + url: str, + headers: dict | None, + body: Any, + timeout: int, + validated_ips: list[str], + *, + save_to_file: bool = False, + include_httpx_metadata: bool = False, + ) -> Data: + """Make the request and follow redirects manually, re-validating every hop. + + This closes an SSRF bypass: with ``follow_redirects`` enabled, httpx would + otherwise auto-follow a redirect from a validated public URL to an internal + address that was never checked. Here each redirect ``Location`` is resolved + (relative locations included) and re-validated with ``validate_and_resolve_url`` + — the same private/loopback/link-local denylist and DNS pinning applied to the + initial request — before any connection to it is made. A blocked hop raises + ``ValueError``; the number of redirects is capped at ``MAX_REDIRECTS``. + """ + method = method.upper() + if method not in {"GET", "POST", "PATCH", "PUT", "DELETE"}: + msg = f"Unsupported method: {method}" + raise ValueError(msg) + + processed_body = self._process_body(body) + current_url = url + current_ips = validated_ips + redirection_history: list[dict] = [] + + for _ in range(MAX_REDIRECTS + 1): + request_params: dict[str, Any] = { + "method": method, + "url": current_url, + "headers": headers, + "timeout": timeout, + # Never let httpx follow redirects itself; each hop is validated below. + "follow_redirects": False, + } + # Only include body for methods that support it (GET must not have a body). + if method in {"POST", "PATCH", "PUT", "DELETE"} and processed_body is not None: + request_params["json"] = processed_body + + try: + async with self._build_http_client(current_url, current_ips) as client: + response = await client.request(**request_params) + except (httpx.HTTPError, httpx.RequestError, httpx.TimeoutException) as exc: + self.log(f"Error making request to {current_url}") + return Data( + data={ + "source": url, + "headers": headers, + "status_code": 500, + "error": str(exc), + **({"redirection_history": redirection_history} if redirection_history else {}), + }, + ) + + location = response.headers.get("Location") + if response.status_code in REDIRECT_STATUS_CODES and location: + # Resolve relative redirects against the current URL. + next_url = urljoin(current_url, location) + redirection_history.append({"url": location, "status_code": response.status_code}) + + # Re-validate the redirect target with the same SSRF denylist + DNS pinning. + # Non-http(s) schemes, private/loopback/link-local hosts, and hostnames that + # resolve to blocked IPs all raise SSRFProtectionError here. + try: + _validated_url, current_ips = validate_and_resolve_url(next_url) + except SSRFProtectionError as e: + msg = f"SSRF Protection: blocked redirect to {next_url}: {e}" + raise ValueError(msg) from e + + method = self._method_for_redirect(method, response.status_code) + headers = self._headers_for_redirect(headers, current_url, next_url) + current_url = next_url + continue + + # Not a redirect (or no Location header) - this is the final response. + return await self._build_response_data( + response, + url, + headers, + redirection_history, + save_to_file=save_to_file, + include_httpx_metadata=include_httpx_metadata, + ) + + msg = f"SSRF Protection: exceeded the maximum of {MAX_REDIRECTS} redirects while requesting {url}" + raise ValueError(msg) + def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict: """Update the build config based on the selected mode.""" if field_name != "mode":