diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json b/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json index 347f54eb47..2503356a6c 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json @@ -1327,7 +1327,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 4 diff --git a/src/backend/base/langflow/initial_setup/starter_projects/News Aggregator.json b/src/backend/base/langflow/initial_setup/starter_projects/News Aggregator.json index 34bcff1da8..7ae3584863 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/News Aggregator.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/News Aggregator.json @@ -1781,7 +1781,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 7 diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Pokédex Agent.json b/src/backend/base/langflow/initial_setup/starter_projects/Pokédex Agent.json index a591bc20f3..07b7363350 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Pokédex Agent.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Pokédex Agent.json @@ -768,7 +768,7 @@ "key": "APIRequest", "legacy": false, "metadata": { - "code_hash": "2af407885294", + "code_hash": "4f329dec9a39", "dependencies": { "dependencies": [ { @@ -886,7 +886,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\nfrom lfx.utils.ssrf_protection import SSRFProtectionError, validate_url_for_ssrf\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 optimized parameter handling.\"\"\"\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 when redirects are enabled\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 # if self.mode == \"cURL\" and self.curl_input:\n # self._build_config = self.parse_curl(self.curl_input, dotdict())\n # # After parsing curl, get the normalized URL\n # url = self._build_config[\"url_input\"][\"value\"]\n\n # Normalize URL before validation\n url = self._normalize_url(url)\n\n # Validate URL\n if not validators.url(url):\n msg = f\"Invalid URL provided: {url}\"\n raise ValueError(msg)\n\n # SSRF Protection: Validate URL to prevent access to internal resources\n # TODO: In next major version (2.0), remove warn_only=True to enforce blocking\n try:\n validate_url_for_ssrf(url, warn_only=True)\n except SSRFProtectionError as e:\n # This will only raise if SSRF protection is enabled and warn_only=False\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n\n # Process query parameters\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\n headers = self._process_headers(headers)\n body = self._process_body(body)\n url = self.add_query_params(url, query_params)\n\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 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, 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" }, "curl_input": { "_input_type": "MultilineInput", diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Portfolio Website Code Generator.json b/src/backend/base/langflow/initial_setup/starter_projects/Portfolio Website Code Generator.json index f66cee1ec5..bbf8fe2f5c 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Portfolio Website Code Generator.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Portfolio Website Code Generator.json @@ -949,7 +949,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 4 diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Text Sentiment Analysis.json b/src/backend/base/langflow/initial_setup/starter_projects/Text Sentiment Analysis.json index e6abf4c19f..64b84fa493 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Text Sentiment Analysis.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Text Sentiment Analysis.json @@ -2641,7 +2641,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 4 diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Vector Store RAG.json b/src/backend/base/langflow/initial_setup/starter_projects/Vector Store RAG.json index 13ee4dee83..7967fed23e 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Vector Store RAG.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Vector Store RAG.json @@ -1643,7 +1643,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 4 diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Youtube Analysis.json b/src/backend/base/langflow/initial_setup/starter_projects/Youtube Analysis.json index c373dc7eb9..5441bdbe3e 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Youtube Analysis.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Youtube Analysis.json @@ -274,7 +274,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "lfx", 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 64834643e5..37d21819c6 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 @@ -363,7 +363,15 @@ class TestAPIRequestComponent(ComponentTestBaseWithoutClient): class TestAPIRequestSSRFProtection: - """Test SSRF protection in API Request component.""" + """Rewritten SSRF Protection Tests for API Request Component. + + These tests properly test the actual SSRF protection implementation without mocking + the core security functions. They verify: + 1. Real SSRF blocking with actual private IPs + 2. DNS pinning actually prevents rebinding + 3. Allowlist functionality works correctly + 4. Custom transport is used when protection is enabled. + """ @pytest.fixture def component_class(self): @@ -376,10 +384,10 @@ class TestAPIRequestSSRFProtection: return { "url_input": "https://example.com/api/test", "method": "GET", - "headers": [{"key": "User-Agent", "value": "test-agent"}], + "headers": [], "body": [], "timeout": 30, - "follow_redirects": False, # Changed default for SSRF security + "follow_redirects": False, "save_to_file": False, "include_httpx_metadata": False, "mode": "URL", @@ -392,73 +400,93 @@ class TestAPIRequestSSRFProtection: """Return a component instance.""" return component_class(**default_kwargs) - async def test_ssrf_protection_disabled_by_default(self, component): - """Test that SSRF protection is disabled by default (warn-only mode).""" - # Even with protection disabled, this should not raise + async def test_ssrf_protection_disabled_allows_all_urls(self, component): + """Test that when SSRF protection is disabled, all URLs are allowed.""" component.url_input = "http://127.0.0.1:8080" - with respx.mock: + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}), + respx.mock, + ): respx.get("http://127.0.0.1:8080").mock(return_value=Response(200, json={"status": "ok"})) - # Should not raise (protection is off by default) result = await component.make_api_request() assert isinstance(result, Data) + assert result.data["result"]["status"] == "ok" - async def test_ssrf_protection_enabled_blocks_localhost(self, component): - """Test that SSRF protection blocks localhost when enabled.""" + async def test_ssrf_protection_blocks_localhost_127_0_0_1(self, component): + """Test that SSRF protection blocks 127.0.0.1 (localhost).""" component.url_input = "http://127.0.0.1:8080/admin" - # Enable SSRF protection in enforcement mode with ( patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), - patch("lfx.components.data_source.api_request.validate_url_for_ssrf") as mock_validate, + pytest.raises(ValueError, match="SSRF Protection"), ): - from lfx.utils.ssrf_protection import SSRFProtectionError + await component.make_api_request() - # Make it raise in enforcement mode - mock_validate.side_effect = SSRFProtectionError("Access to 127.0.0.1 blocked") + async def test_ssrf_protection_blocks_localhost_0_0_0_0(self, component): + """Test that SSRF protection blocks 0.0.0.0.""" + component.url_input = "http://0.0.0.0:8080/admin" - with pytest.raises(ValueError, match="SSRF Protection"): - await component.make_api_request() + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="SSRF Protection"), + ): + await component.make_api_request() - async def test_ssrf_protection_enabled_blocks_private_networks(self, component): - """Test that SSRF protection blocks private network IPs when enabled.""" - private_ips = [ - "http://192.168.1.1/config", - "http://10.0.0.1/admin", - "http://172.16.0.1/internal", - ] + async def test_ssrf_protection_blocks_private_network_192_168(self, component): + """Test that SSRF protection blocks 192.168.x.x private network.""" + component.url_input = "http://192.168.1.1/config" - with patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}): - for url in private_ips: - component.url_input = url + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="SSRF Protection"), + ): + await component.make_api_request() - with patch("lfx.components.data_source.api_request.validate_url_for_ssrf") as mock_validate: - from lfx.utils.ssrf_protection import SSRFProtectionError + async def test_ssrf_protection_blocks_private_network_10_0(self, component): + """Test that SSRF protection blocks 10.x.x.x private network.""" + component.url_input = "http://10.0.0.1/admin" - mock_validate.side_effect = SSRFProtectionError(f"Access to {url} blocked") + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="SSRF Protection"), + ): + await component.make_api_request() - with pytest.raises(ValueError, match="SSRF Protection"): - await component.make_api_request() + async def test_ssrf_protection_blocks_private_network_172_16(self, component): + """Test that SSRF protection blocks 172.16.x.x private network.""" + component.url_input = "http://172.16.0.1/internal" - async def test_ssrf_protection_enabled_blocks_metadata_endpoint(self, component): - """Test that SSRF protection blocks cloud metadata endpoints when enabled.""" + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="SSRF Protection"), + ): + await component.make_api_request() + + async def test_ssrf_protection_blocks_cloud_metadata_endpoint(self, component): + """Test that SSRF protection blocks AWS/GCP metadata endpoint.""" component.url_input = "http://169.254.169.254/latest/meta-data/" with ( patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), - patch("lfx.components.data_source.api_request.validate_url_for_ssrf") as mock_validate, + pytest.raises(ValueError, match="SSRF Protection"), ): - from lfx.utils.ssrf_protection import SSRFProtectionError + await component.make_api_request() - mock_validate.side_effect = SSRFProtectionError("Access to 169.254.169.254 blocked") + async def test_ssrf_protection_blocks_link_local_169_254(self, component): + """Test that SSRF protection blocks link-local addresses.""" + component.url_input = "http://169.254.1.1/api" - with pytest.raises(ValueError, match="SSRF Protection"): - await component.make_api_request() + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + pytest.raises(ValueError, match="SSRF Protection"), + ): + await component.make_api_request() @respx.mock async def test_ssrf_protection_allows_public_urls(self, component): - """Test that SSRF protection allows public URLs.""" + """Test that SSRF protection allows legitimate public URLs.""" public_urls = [ "https://api.openai.com/v1/chat/completions", "https://api.github.com/repos/langflow-ai/langflow", @@ -470,56 +498,133 @@ class TestAPIRequestSSRFProtection: component.url_input = url respx.get(url).mock(return_value=Response(200, json={"status": "ok"})) - # Should not raise - these are public URLs result = await component.make_api_request() assert isinstance(result, Data) + assert result.data["result"]["status"] == "ok" - async def test_ssrf_protection_allowlist_bypass(self, component): - """Test that allowlisted hosts bypass SSRF protection.""" + async def test_ssrf_allowlist_hostname(self, component): + """Test that allowlisted hostnames bypass SSRF protection.""" component.url_input = "http://internal.company.local/api" with ( patch.dict( os.environ, - {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true", "LANGFLOW_SSRF_ALLOWED_HOSTS": "internal.company.local"}, + { + "LANGFLOW_SSRF_PROTECTION_ENABLED": "true", + "LANGFLOW_SSRF_ALLOWED_HOSTS": "internal.company.local", + }, ), respx.mock, ): respx.get("http://internal.company.local/api").mock(return_value=Response(200, json={"status": "ok"})) - # Should not raise - host is in allowlist result = await component.make_api_request() assert isinstance(result, Data) + assert result.data["result"]["status"] == "ok" - async def test_ssrf_protection_allowlist_cidr(self, component): + async def test_ssrf_allowlist_ip_address(self, component): + """Test that allowlisted IP addresses bypass SSRF protection.""" + component.url_input = "http://192.168.1.100/api" + + with ( + patch.dict( + os.environ, + { + "LANGFLOW_SSRF_PROTECTION_ENABLED": "true", + "LANGFLOW_SSRF_ALLOWED_HOSTS": "192.168.1.100", + }, + ), + respx.mock, + ): + respx.get("http://192.168.1.100/api").mock(return_value=Response(200, json={"status": "ok"})) + + result = await component.make_api_request() + assert isinstance(result, Data) + assert result.data["result"]["status"] == "ok" + + async def test_ssrf_allowlist_cidr_range(self, component): """Test that CIDR ranges in allowlist work correctly.""" component.url_input = "http://192.168.1.5/api" with ( patch.dict( os.environ, - {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true", "LANGFLOW_SSRF_ALLOWED_HOSTS": "192.168.1.0/24"}, + { + "LANGFLOW_SSRF_PROTECTION_ENABLED": "true", + "LANGFLOW_SSRF_ALLOWED_HOSTS": "192.168.1.0/24", + }, + ), + respx.mock, + ): + respx.get("http://192.168.1.5/api").mock(return_value=Response(200, json={"status": "ok"})) + + result = await component.make_api_request() + assert isinstance(result, Data) + assert result.data["result"]["status"] == "ok" + + async def test_ssrf_allowlist_multiple_entries(self, component): + """Test that multiple allowlist entries work correctly.""" + component.url_input = "http://192.168.1.5/api" + + with ( + patch.dict( + os.environ, + { + "LANGFLOW_SSRF_PROTECTION_ENABLED": "true", + "LANGFLOW_SSRF_ALLOWED_HOSTS": "localhost,192.168.1.0/24,internal.local", + }, ), respx.mock, ): respx.get("http://192.168.1.5/api").mock(return_value=Response(200, json={"status": "ok"})) - # Should not raise - IP is in allowlisted CIDR range result = await component.make_api_request() assert isinstance(result, Data) - async def test_ssrf_protection_warn_only_mode(self, component): - """Test that warn_only mode logs warnings instead of blocking.""" - component.url_input = "http://127.0.0.1:8080/admin" + async def test_dns_pinning_is_used_when_protection_enabled(self, component): + """Test that DNS pinning (custom transport) is used when SSRF protection is enabled.""" + from unittest.mock import AsyncMock - with patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), respx.mock: - respx.get("http://127.0.0.1:8080/admin").mock(return_value=Response(200, json={"status": "ok"})) + component.url_input = "https://example.com/api" + + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + patch("lfx.components.data_source.api_request.create_ssrf_protected_client") as mock_create_client, + respx.mock, + ): + # Mock the context manager returned by create_ssrf_protected_client + mock_client = AsyncMock() + mock_client.__aenter__.return_value = mock_client + mock_client.__aexit__.return_value = None + mock_create_client.return_value = mock_client + + # Mock the make_request to return a Data object + component.make_request = AsyncMock(return_value=Data(data={"status": "ok"})) + + await component.make_api_request() + + # Verify that create_ssrf_protected_client was called (DNS pinning is used) + mock_create_client.assert_called_once() + call_kwargs = mock_create_client.call_args[1] + assert call_kwargs["hostname"] == "example.com" + assert len(call_kwargs["validated_ips"]) > 0 # Should have validated IPs + + async def test_normal_client_used_when_protection_disabled(self, component): + """Test that normal httpx client is used when SSRF protection is disabled.""" + component.url_input = "https://example.com/api" + + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}), + patch("lfx.components.data_source.api_request.create_ssrf_protected_client") as mock_create_client, + respx.mock, + ): + respx.get("https://example.com/api").mock(return_value=Response(200, json={"status": "ok"})) - # In warn_only mode (default), should not raise but should log result = await component.make_api_request() - assert isinstance(result, Data) - # TODO: In next major version, this should raise instead of just warning + # Verify that create_ssrf_protected_client was NOT called + mock_create_client.assert_not_called() + assert isinstance(result, Data) async def test_follow_redirects_security_warning(self, component): """Test that enabling follow_redirects logs a security warning.""" @@ -527,58 +632,55 @@ class TestAPIRequestSSRFProtection: component.url_input = "https://example.com/api" component.follow_redirects = True - - # Mock the log method to capture what's being logged component.log = MagicMock() - with respx.mock: + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}), + respx.mock, + ): respx.get("https://example.com/api").mock(return_value=Response(200, json={"status": "ok"})) - result = await component.make_api_request() - assert isinstance(result, Data) + await component.make_api_request() - # Verify log was called with security warning + # Verify security warning was logged component.log.assert_called() - log_call_args = component.log.call_args[0][0] - assert "Security Warning" in log_call_args - assert "SSRF bypass" in log_call_args - assert "redirects are enabled" in log_call_args + all_log_messages = [call[0][0] for call in component.log.call_args_list] + security_warning_found = any("Security Warning" in msg and "SSRF bypass" in msg for msg in all_log_messages) + assert security_warning_found, f"Security warning not found in: {all_log_messages}" - async def test_follow_redirects_disabled_by_default(self, component): - """Test that follow_redirects is disabled by default.""" - # Verify the default value is False - assert component.follow_redirects is False - - async def test_url_normalization(self, component): + async def test_url_normalization_adds_https(self, component): """Test that URLs without protocol get normalized to https://.""" - # Test URL without protocol component.url_input = "example.com" - with respx.mock: + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}), + respx.mock, + ): respx.get("https://example.com").mock(return_value=Response(200, json={"status": "ok"})) result = await component.make_api_request() - assert isinstance(result, Data) assert result.data["source"] == "https://example.com" - async def test_url_normalization_preserves_protocol(self, component): - """Test that URLs with protocol are not modified.""" - # Test http:// is preserved + async def test_url_normalization_preserves_http(self, component): + """Test that http:// protocol is preserved.""" component.url_input = "http://example.com" - with respx.mock: + with ( + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}), + respx.mock, + ): respx.get("http://example.com").mock(return_value=Response(200, json={"status": "ok"})) result = await component.make_api_request() - assert isinstance(result, Data) assert result.data["source"] == "http://example.com" - # Test https:// is preserved - component.url_input = "https://example.com" + async def test_invalid_url_raises_error(self, component): + """Test that invalid URLs raise ValueError.""" + component.url_input = "not_a_valid_url" - with respx.mock: - respx.get("https://example.com").mock(return_value=Response(200, json={"status": "ok"})) + with pytest.raises(ValueError, match="Invalid URL provided"): + await component.make_api_request() - result = await component.make_api_request() - assert isinstance(result, Data) - assert result.data["source"] == "https://example.com" + 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 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 new file mode 100644 index 0000000000..5471b685f1 --- /dev/null +++ b/src/backend/tests/unit/components/data_source/test_dns_rebinding.py @@ -0,0 +1,374 @@ +"""Test DNS rebinding protection in API Request component. + +This test suite verifies that the DNS pinning implementation prevents +DNS rebinding attacks that could bypass SSRF protection. +""" +# ruff: noqa: ARG001, SIM117 + +import os +import socket +from unittest.mock import patch + +import httpcore +import httpx +import pytest +from lfx.components.data_source.api_request import APIRequestComponent +from lfx.schema import Data + + +class TestDNSRebindingProtection: + """Test DNS rebinding attack prevention through DNS pinning.""" + + @pytest.fixture + def component(self): + """Create a basic API request component.""" + return APIRequestComponent( + url_input="http://rebinding.test:8080/api", + method="GET", + headers=[], + body=[], + timeout=30, + follow_redirects=False, + save_to_file=False, + include_httpx_metadata=False, + mode="URL", + curl_input="", + query_params={}, + ) + + @pytest.mark.asyncio + async def test_dns_pinning_prevents_rebinding_attack(self, component): + """Test that DNS pinning prevents DNS rebinding attacks. + + This test simulates a DNS rebinding attack where: + 1. First DNS lookup (validation): returns public IP (8.8.8.8) + 2. Second DNS lookup (httpx): would return localhost (127.0.0.1) + + With DNS pinning, the second lookup should NOT happen - the validated + IP from the first lookup should be used directly at the network layer. + """ + call_count = 0 + connected_to_ip = None + + def mock_getaddrinfo(_hostname, _port, *_args, **_kwargs): + """Mock DNS resolution to simulate rebinding attack.""" + nonlocal call_count + call_count += 1 + + if call_count == 1: + # First call (during validation): return public IP + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 0))] + # Second call (during httpx request): return localhost + # This simulates the DNS rebinding attack + # With DNS pinning, this should NOT be called + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 0))] + + # Mock the network backend's connect_tcp to capture the actual IP being connected to + async def mock_connect_tcp(self, host, port, **kwargs): + """Capture the IP that's actually being connected to.""" + nonlocal connected_to_ip + connected_to_ip = host + # Return a mock stream with proper format (list of bytes) + return httpcore.AsyncMockStream( + [ + 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"}', + ] + ) + + 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), + ): + # Execute the request + result = await component.make_api_request() + + # Verify the request succeeded + assert result is not None + + # CRITICAL CHECK: DNS should only be called once (during validation) + # If called twice, DNS pinning failed and the attack succeeded + assert call_count == 1, ( + f"DNS was called {call_count} times. Expected 1 (validation only). " + "DNS pinning failed - the component is vulnerable to DNS rebinding!" + ) + + # Verify the connection was made to the pinned IP + assert connected_to_ip is not None, "No TCP connection was made" + assert connected_to_ip == "8.8.8.8", ( + f"Connection should be to pinned IP 8.8.8.8, but was to {connected_to_ip}" + ) + + @pytest.mark.asyncio + async def test_dns_pinning_preserves_hostname_in_header(self, component): + """Test that DNS pinning connects to pinned IP while preserving hostname for TLS. + + With network-level DNS pinning: + - TCP connection goes to the pinned IP (93.184.216.34) + - URL preserves the original hostname (rebinding.test) + - This allows TLS SNI and certificate verification to work correctly + + This is important for: + - Virtual hosting (multiple sites on same IP) + - SNI (Server Name Indication) for HTTPS + - Certificate verification (cert is for hostname, not IP) + """ + connected_to_ip = None + + def mock_getaddrinfo(hostname, port, *args, **kwargs): + """Mock DNS resolution.""" + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + + # Mock network backend to capture TCP connection + async def mock_connect_tcp(self, host, port, **kwargs): + """Capture the IP that TCP connects to.""" + nonlocal connected_to_ip + connected_to_ip = host + return httpcore.AsyncMockStream( + [ + 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"}', + ] + ) + + 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() + + # Verify the request succeeded + assert result is not None + + # Verify TCP connection was made to the pinned IP + assert connected_to_ip is not None, "No TCP connection was made" + assert connected_to_ip == "93.184.216.34", ( + f"TCP connection should be to pinned IP 93.184.216.34, but was to {connected_to_ip}" + ) + + @pytest.mark.asyncio + async def test_dns_pinning_with_direct_ip_address(self, component): + """Test that direct IP addresses work correctly (no DNS pinning needed).""" + component.url_input = "http://93.184.216.34:8080/api" + + def mock_getaddrinfo(hostname, port, *args, **kwargs): + """Mock DNS resolution - should not be called for direct IPs.""" + # For direct IPs, socket.getaddrinfo is still called but returns the same IP + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + + mock_response = httpx.Response(200, json={"status": "ok"}) + + with ( + patch("socket.getaddrinfo", side_effect=mock_getaddrinfo), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + patch("httpx.AsyncClient.request", return_value=mock_response) as mock_request, + ): + result = await component.make_api_request() + + # Verify the request succeeded + assert isinstance(result, Data) + assert mock_request.called + + @pytest.mark.asyncio + async def test_dns_pinning_disabled_when_protection_disabled(self, component): + """Test that DNS pinning is skipped when SSRF protection is disabled.""" + call_count = 0 + + def mock_getaddrinfo(hostname, port, *args, **kwargs): + """Mock DNS resolution.""" + nonlocal call_count + call_count += 1 + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 0))] + + mock_response = httpx.Response(200, json={"status": "ok"}) + + with ( + patch("socket.getaddrinfo", side_effect=mock_getaddrinfo), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "false"}), + patch("httpx.AsyncClient.request", return_value=mock_response) as mock_request, + ): + result = await component.make_api_request() + + # Verify the request succeeded + assert isinstance(result, Data) + assert mock_request.called + + # When protection is disabled, DNS pinning is not used + # So DNS might be called multiple times (once by httpx) + # This is expected behavior when protection is off + + @pytest.mark.asyncio + async def test_dns_pinning_blocks_private_ip_resolution(self, component): + """Test that DNS pinning blocks hostnames that resolve to private IPs.""" + component.url_input = "http://internal.example.com/api" + + def mock_getaddrinfo(hostname, port, *args, **kwargs): + """Mock DNS resolution to return private IP.""" + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.168.1.1", 0))] + + with ( + patch("socket.getaddrinfo", side_effect=mock_getaddrinfo), + patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}), + ): + # Should raise ValueError due to SSRF protection + with pytest.raises(ValueError, match="SSRF Protection"): + await component.make_api_request() + + @pytest.mark.asyncio + async def test_dns_pinning_with_ipv6(self, component): + """Test that DNS pinning works with IPv6 addresses. + + With network-level DNS pinning: + - TCP connection goes to the IPv6 address + - URL preserves the original hostname + - IPv6 addresses don't need brackets in connect_tcp (only in URLs) + """ + component.url_input = "http://ipv6.example.com/api" + connected_to_ip = None + + def mock_getaddrinfo(hostname, port, *args, **kwargs): + """Mock DNS resolution to return IPv6.""" + return [(socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2001:4860:4860::8888", 0))] + + # Mock network backend to capture TCP connection + async def mock_connect_tcp(self, host, port, **kwargs): + """Capture the IP that TCP connects to.""" + nonlocal connected_to_ip + connected_to_ip = host + return httpcore.AsyncMockStream( + [ + 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"}', + ] + ) + + 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() + + # Verify the request succeeded + assert isinstance(result, Data) + + # Verify TCP connection was made to the IPv6 address + assert connected_to_ip is not None, "No TCP connection was made" + assert connected_to_ip == "2001:4860:4860::8888", ( + f"TCP connection should be to IPv6 2001:4860:4860::8888, but was to {connected_to_ip}" + ) + + @pytest.mark.asyncio + async def test_dns_pinning_with_allowlisted_host(self, component): + """Test that allowlisted hosts bypass DNS pinning and preserve original hostname.""" + component.url_input = "http://internal.example.com:8080/api" # Use a valid hostname format + captured_request = None + + def mock_getaddrinfo(hostname, port, *args, **kwargs): + """Mock DNS resolution.""" + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.1", 0))] + + # Mock the transport's handle_async_request to capture the rewritten request + async def mock_handle_async_request(_self, request): + """Capture the request after transport rewrite.""" + nonlocal captured_request + captured_request = request + return httpx.Response(200, json={"status": "ok"}, request=request) + + with ( + patch("socket.getaddrinfo", side_effect=mock_getaddrinfo), + patch.dict( + os.environ, + { + "LANGFLOW_SSRF_PROTECTION_ENABLED": "true", + "LANGFLOW_SSRF_ALLOWED_HOSTS": "internal.example.com,10.0.0.1", + }, + ), + # Patch get_allowed_hosts to return the allowlist directly + patch( + "lfx.utils.ssrf_protection.get_allowed_hosts", + return_value=["internal.example.com", "10.0.0.1"], + ), + patch.object(httpx.AsyncHTTPTransport, "handle_async_request", mock_handle_async_request), + ): + result = await component.make_api_request() + + # Verify the request succeeded + assert isinstance(result, Data) + assert captured_request is not None, "Transport did not capture request" + + # Verify the original hostname is preserved (no DNS pinning for allowlisted hosts) + url_str = str(captured_request.url) + assert "internal.example.com" in url_str, f"Allowlisted host should preserve hostname: {url_str}" + assert "10.0.0.1" not in url_str, f"Allowlisted host should not use IP: {url_str}" + + @pytest.mark.asyncio + async def test_dns_pinning_with_multiple_ips_fallback(self, component): + """Test that DNS pinning tries multiple IPs when first one fails (dual-stack/load balancing).""" + component.url_input = "http://dual-stack.example.com/api" + + # Track which IPs were attempted + attempted_ips = [] + + def mock_getaddrinfo(host, port, family=0, type_=0, proto=0, flags=0): + """Mock DNS resolution to return multiple IPs (IPv4 and IPv6).""" + if host == "dual-stack.example.com": + # Return both IPv4 and IPv6 addresses (dual-stack) + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)), # IPv4 + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2606:2800:220:1:248:1893:25c8:1946", 0)), # IPv6 + ] + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (host, 0))] + + async def mock_connect_tcp(self, host, port, **kwargs): + """Mock connection that fails on first IP, succeeds on second.""" + nonlocal attempted_ips + attempted_ips.append(host) + + # First IP fails (simulating IPv4 unreachable) + if host == "93.184.216.34": + msg = "Connection refused" + raise OSError(msg) + + # Second IP succeeds (IPv6 works) + if host == "2606:2800:220:1:248:1893:25c8:1946": + return httpcore.AsyncMockStream( + [ + 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"}', + ] + ) + + msg = f"Unexpected host: {host}" + raise OSError(msg) + + 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), + ): + # Execute the request + result = await component.make_api_request() + + # Verify both IPs were attempted in order + assert len(attempted_ips) == 2 + assert attempted_ips[0] == "93.184.216.34" # IPv4 tried first + assert attempted_ips[1] == "2606:2800:220:1:248:1893:25c8:1946" # IPv6 tried second + + # Verify the result (should succeed with second IP) + assert isinstance(result, Data) + assert result.data is not None diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index d85ba87cd6..3bac0a9c44 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -3253,7 +3253,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 5 @@ -54090,7 +54090,7 @@ }, { "name": "couchbase", - "version": "4.6.0" + "version": "4.6.1" } ], "total_dependencies": 3 @@ -56895,7 +56895,7 @@ "icon": "Globe", "legacy": false, "metadata": { - "code_hash": "2af407885294", + "code_hash": "4f329dec9a39", "dependencies": { "dependencies": [ { @@ -56997,7 +56997,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\nfrom lfx.utils.ssrf_protection import SSRFProtectionError, validate_url_for_ssrf\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 optimized parameter handling.\"\"\"\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 when redirects are enabled\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 # if self.mode == \"cURL\" and self.curl_input:\n # self._build_config = self.parse_curl(self.curl_input, dotdict())\n # # After parsing curl, get the normalized URL\n # url = self._build_config[\"url_input\"][\"value\"]\n\n # Normalize URL before validation\n url = self._normalize_url(url)\n\n # Validate URL\n if not validators.url(url):\n msg = f\"Invalid URL provided: {url}\"\n raise ValueError(msg)\n\n # SSRF Protection: Validate URL to prevent access to internal resources\n # TODO: In next major version (2.0), remove warn_only=True to enforce blocking\n try:\n validate_url_for_ssrf(url, warn_only=True)\n except SSRFProtectionError as e:\n # This will only raise if SSRF protection is enabled and warn_only=False\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n\n # Process query parameters\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\n headers = self._process_headers(headers)\n body = self._process_body(body)\n url = self.add_query_params(url, query_params)\n\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 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, 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" }, "curl_input": { "_input_type": "MultilineInput", @@ -63838,7 +63838,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 6 @@ -68214,7 +68214,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 4 @@ -69517,7 +69517,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" } ], "total_dependencies": 7 @@ -72150,7 +72150,7 @@ }, { "name": "git", - "version": "3.1.43" + "version": "3.1.47" }, { "name": "lfx", @@ -72907,7 +72907,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "langchain_core", @@ -73322,7 +73322,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "lfx", @@ -81279,7 +81279,7 @@ }, { "name": "langchain", - "version": "1.2.15" + "version": "1.2.17" } ], "total_dependencies": 3 @@ -86257,7 +86257,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 5 @@ -88527,7 +88527,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 4 @@ -92615,7 +92615,7 @@ "dependencies": [ { "name": "toolguard", - "version": "0.2.16" + "version": "0.2.17" }, { "name": "lfx", @@ -97547,7 +97547,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 4 @@ -113454,7 +113454,7 @@ }, { "name": "langchain_google_vertexai", - "version": "3.2.2" + "version": "3.2.3" }, { "name": "google", @@ -113838,7 +113838,7 @@ }, { "name": "langchain_google_vertexai", - "version": "3.2.2" + "version": "3.2.3" }, { "name": "google", @@ -114586,7 +114586,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 4 @@ -116011,7 +116011,7 @@ }, { "name": "openai", - "version": "2.32.0" + "version": "2.33.0" } ], "total_dependencies": 6 @@ -116563,7 +116563,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "lfx", @@ -116752,7 +116752,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "lfx", @@ -117068,7 +117068,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "lfx", @@ -117467,7 +117467,7 @@ }, { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "lfx", @@ -117730,7 +117730,7 @@ "dependencies": [ { "name": "googleapiclient", - "version": "2.194.0" + "version": "2.195.0" }, { "name": "pandas", @@ -118101,6 +118101,6 @@ "num_components": 355, "num_modules": 97 }, - "sha256": "30e1190dc06f5782586a9628ddb7a5705cd1aa49d4d47e90503a5ddaf85290af", + "sha256": "3a101632bdf455f8b0936f73acc5a1ae07deed3c2909a49465b85839f8e4f0d5", "version": "0.4.2" } 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 e599d6a406..85350dc581 100644 --- a/src/lfx/src/lfx/components/data_source/api_request.py +++ b/src/lfx/src/lfx/components/data_source/api_request.py @@ -27,7 +27,10 @@ from lfx.io import ( from lfx.schema.data import Data from lfx.schema.dotdict import dotdict from lfx.utils.component_utils import set_current_fields, set_field_advanced, set_field_display -from lfx.utils.ssrf_protection import SSRFProtectionError, validate_url_for_ssrf + +# 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_transport import create_ssrf_protected_client # Define fields for each mode MODE_FIELDS = { @@ -422,7 +425,22 @@ class APIRequestComponent(Component): return {} async def make_api_request(self) -> Data: - """Make HTTP request with optimized parameter handling.""" + """Make HTTP request with SSRF protection and DNS pinning. + + This method implements comprehensive SSRF (Server-Side Request Forgery) protection + using DNS pinning to prevent DNS rebinding attacks. The protection works by: + 1. Validating the URL and resolving DNS during security check + 2. Pinning the validated IP address + 3. Forcing the HTTP client to use the pinned IP for the actual request + 4. Ignoring any subsequent DNS changes (prevents rebinding attacks) + + Returns: + Data: Response data from the HTTP request + + Raises: + ValueError: If URL is invalid or blocked by SSRF protection + """ + # Extract request parameters method = self.method url = self.url_input.strip() if isinstance(self.url_input, str) else "" headers = self.headers or {} @@ -432,7 +450,8 @@ class APIRequestComponent(Component): save_to_file = self.save_to_file include_httpx_metadata = self.include_httpx_metadata - # Security warning when redirects are enabled + # Security warning: HTTP redirects can bypass SSRF protection + # A public URL could redirect to an internal resource if follow_redirects: self.log( "Security Warning: HTTP redirects are enabled. This may allow SSRF bypass attacks " @@ -440,51 +459,122 @@ class APIRequestComponent(Component): "Only enable this if you trust the target server." ) - # if self.mode == "cURL" and self.curl_input: - # self._build_config = self.parse_curl(self.curl_input, dotdict()) - # # After parsing curl, get the normalized URL - # url = self._build_config["url_input"]["value"] - - # Normalize URL before validation + # Normalize URL (add https:// if no protocol specified) url = self._normalize_url(url) - # Validate URL + # Basic URL format validation if not validators.url(url): msg = f"Invalid URL provided: {url}" raise ValueError(msg) - # SSRF Protection: Validate URL to prevent access to internal resources - # TODO: In next major version (2.0), remove warn_only=True to enforce blocking + # ============================================================================ + # SSRF Protection with DNS Pinning + # ============================================================================ + # This prevents DNS rebinding attacks by: + # 1. Resolving DNS and validating IPs during security check + # 2. Pinning the validated IP address + # 3. Using a custom HTTP transport that forces use of the pinned IP + # 4. Ignoring any new DNS resolutions (prevents rebinding) + # + # Without DNS pinning, an attacker could: + # - First DNS lookup: returns public IP (passes validation) + # - Second DNS lookup: returns internal IP (bypasses protection) + # - Attack succeeds: accesses internal services + # + # With DNS pinning: + # - First DNS lookup: returns public IP (passes validation) + # - IP is pinned: "example.com = 93.184.216.34" + # - HTTP request: uses pinned IP directly (no new DNS lookup) + # - Attack fails: even if DNS changes, we use the validated IP + # ============================================================================ + try: - validate_url_for_ssrf(url, warn_only=True) + # Validate URL and get validated IPs for DNS pinning + _validated_url, validated_ips = validate_and_resolve_url(url) + + # Log DNS pinning information for security auditing + if validated_ips: + self.log(f"SSRF Protection: Using DNS pinning with {len(validated_ips)} validated IP(s)") + except SSRFProtectionError as e: - # This will only raise if SSRF protection is enabled and warn_only=False + # SSRF protection blocked the request (private IP, internal network, etc.) msg = f"SSRF Protection: {e}" raise ValueError(msg) from e - # Process query parameters + # Process query parameters (from string or Data object) if isinstance(self.query_params, str): query_params = dict(parse_qsl(self.query_params)) else: query_params = self.query_params.data if self.query_params else {} - # Process headers and body + # Process headers and body into proper format headers = self._process_headers(headers) body = self._process_body(body) url = self.add_query_params(url, query_params) - 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, - ) + # ============================================================================ + # Create HTTP Client with DNS Pinning (if SSRF protection enabled) + # ============================================================================ + 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, + ) + 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: + 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, + ) + self.status = result return result diff --git a/src/lfx/src/lfx/services/settings/base.py b/src/lfx/src/lfx/services/settings/base.py index f00eac3512..3828cb6228 100644 --- a/src/lfx/src/lfx/services/settings/base.py +++ b/src/lfx/src/lfx/services/settings/base.py @@ -398,12 +398,12 @@ class Settings(BaseSettings): use hardware-level isolation to restrict access.""" # SSRF Protection - ssrf_protection_enabled: bool = False + ssrf_protection_enabled: bool = True """If set to True, Langflow will enable SSRF (Server-Side Request Forgery) protection. When enabled, blocks requests to private IP ranges, localhost, and cloud metadata endpoints. - When False (default), no URL validation is performed, allowing requests to any destination + When False, no URL validation is performed, allowing requests to any destination including internal services, private networks, and cloud metadata endpoints. - Default is False for backward compatibility. In v2.0, this will be changed to True. + Default is True to protect against SSRF attacks including DNS rebinding. Note: When ssrf_protection_enabled is disabled, the ssrf_allowed_hosts setting is ignored and has no effect.""" ssrf_allowed_hosts: list[str] = [] diff --git a/src/lfx/src/lfx/utils/ssrf_protection.py b/src/lfx/src/lfx/utils/ssrf_protection.py index d554055b3f..4044398d5f 100644 --- a/src/lfx/src/lfx/utils/ssrf_protection.py +++ b/src/lfx/src/lfx/utils/ssrf_protection.py @@ -13,15 +13,9 @@ IMPORTANT: HTTP Redirects See: https://cheatsheetseries.owasp.org/cheatsheets/Server_Side_Request_Forgery_Prevention_Cheat_Sheet.html Configuration: - LANGFLOW_SSRF_PROTECTION_ENABLED: Enable/disable SSRF protection (default: false) - TODO: Change default to true in next major version (2.0) + LANGFLOW_SSRF_PROTECTION_ENABLED: Enable/disable SSRF protection (default: true) LANGFLOW_SSRF_ALLOWED_HOSTS: Comma-separated list of allowed hosts/CIDR ranges Examples: "192.168.1.0/24,internal-api.company.local,10.0.0.5" - -TODO: In next major version (2.0): - - Change LANGFLOW_SSRF_PROTECTION_ENABLED default to "true" - - Remove warning-only mode and enforce blocking - - Update documentation to reflect breaking change """ import functools @@ -83,6 +77,16 @@ def is_ssrf_protection_enabled() -> bool: Returns: bool: True if SSRF protection is enabled, False otherwise. """ + # Read directly from environment variable to support test mocking with patch.dict() + # This ensures tests can override the protection state without settings service caching issues + import os + + env_value = os.getenv("LANGFLOW_SSRF_PROTECTION_ENABLED") + if env_value is not None: + # Environment variable is set - use it (supports test mocking) + return env_value.lower() in ("true", "1", "yes", "on") + + # Fall back to settings service for non-test scenarios return get_settings_service().settings.ssrf_protection_enabled @@ -92,11 +96,23 @@ def get_allowed_hosts() -> list[str]: Returns: list[str]: Stripped hostnames or CIDR blocks from settings, or empty list if unset. """ - allowed_hosts = get_settings_service().settings.ssrf_allowed_hosts - if not allowed_hosts: - return [] - # ssrf_allowed_hosts is already a list[str], just clean and filter entries - return [host.strip() for host in allowed_hosts if host and host.strip()] + # Read directly from environment variable to support test mocking with patch.dict() + # This ensures tests can override the allowlist without settings service caching issues + import os + + env_value = os.getenv("LANGFLOW_SSRF_ALLOWED_HOSTS", "") + if env_value: + # Parse comma-separated list from environment variable + return [host.strip() for host in env_value.split(",") if host.strip()] + + # Fall back to settings service for non-test scenarios + settings_service = get_settings_service() + if settings_service: + allowed_hosts = settings_service.settings.ssrf_allowed_hosts + if allowed_hosts: + return [host.strip() for host in allowed_hosts if host and host.strip()] + + return [] def is_host_allowed(hostname: str, ip: str | None = None) -> bool: @@ -271,7 +287,6 @@ def _validate_direct_ip_address(hostname: str) -> bool: if is_ip_blocked(ip_obj): msg = ( f"Access to IP address {hostname} is blocked by SSRF protection. " - "Requests to private/internal IP ranges are not allowed for security reasons. " "To allow this IP, add it to LANGFLOW_SSRF_ALLOWED_HOSTS environment variable." ) raise SSRFProtectionError(msg) @@ -313,15 +328,12 @@ def _validate_hostname_resolution(hostname: str) -> None: if blocked_ips: msg = ( f"Hostname {hostname} resolves to blocked IP address(es): {', '.join(blocked_ips)}. " - "Requests to private/internal IP ranges are not allowed for security reasons. " - "This protection prevents access to internal services, cloud metadata endpoints " - "(e.g., AWS 169.254.169.254), and other sensitive internal resources. " "To allow this hostname, add it to LANGFLOW_SSRF_ALLOWED_HOSTS environment variable." ) raise SSRFProtectionError(msg) -def validate_url_for_ssrf(url: str, *, warn_only: bool = True) -> None: +def validate_url_for_ssrf(url: str, *, warn_only: bool = False) -> None: """Validate a URL to prevent SSRF attacks. This function performs the following checks: @@ -333,8 +345,7 @@ def validate_url_for_ssrf(url: str, *, warn_only: bool = True) -> None: Args: url: URL to validate - warn_only: If True, only log warnings instead of raising errors (default: True) - TODO: Change default to False in next major version (2.0) + warn_only: If True, only log warnings instead of raising errors (default: False) Raises: SSRFProtectionError: If the URL is blocked due to SSRF protection (only if warn_only=False) @@ -382,3 +393,187 @@ def validate_url_for_ssrf(url: str, *, warn_only: bool = True) -> None: ) return raise + + +def validate_and_resolve_url(url: str) -> tuple[str, list[str]]: + """Validate URL for SSRF and return validated IP addresses for DNS pinning. + + This function is the core of DNS pinning-based SSRF protection. It performs + comprehensive validation and returns the validated IP addresses that should + be used for the actual HTTP request, preventing DNS rebinding attacks. + + DNS Rebinding Attack Prevention: + Without DNS pinning, an attacker can exploit the time gap between validation + and the actual HTTP request: + 1. Validation: DNS returns public IP (8.8.8.8) → passes security check + 2. [Attacker changes DNS with TTL=0] + 3. HTTP request: DNS returns internal IP (127.0.0.1) → bypasses protection + + With DNS pinning (this function): + 1. Validation: DNS returns public IP (8.8.8.8) → passes security check + 2. Function returns: (url, ['8.8.8.8']) → IP is pinned + 3. HTTP request: Uses pinned IP directly → no new DNS lookup → secure + + Args: + url: URL to validate (e.g., "http://example.com/api") + + Returns: + Tuple of (original_url, list_of_validated_ips): + - original_url: The input URL unchanged + - list_of_validated_ips: List of validated IP addresses to use for DNS pinning + Returns empty list if: + - SSRF protection is disabled + - Host is in the allowlist (e.g., localhost for Ollama) + - URL scheme is not http/https + + Raises: + SSRFProtectionError: If URL is blocked by SSRF protection + ValueError: If URL format is invalid + + Example: + >>> # Public domain - returns validated IPs for pinning + >>> url, ips = validate_and_resolve_url("http://example.com") + >>> print(ips) # ['93.184.216.34'] + + >>> # Localhost (if in allowlist) - returns empty list (no pinning needed) + >>> url, ips = validate_and_resolve_url("http://localhost:8080") + >>> print(ips) # [] + + >>> # Private IP - raises SSRFProtectionError + >>> url, ips = validate_and_resolve_url("http://192.168.1.1") + # Raises: SSRFProtectionError("Access to IP address 192.168.1.1 is blocked...") + """ + # ============================================================================ + # Step 1: Check if SSRF protection is enabled + # ============================================================================ + if not is_ssrf_protection_enabled(): + # Protection is disabled - return empty list (no DNS pinning) + return url, [] + + # ============================================================================ + # Step 2: Parse and validate URL format + # ============================================================================ + try: + parsed = urlparse(url) + except Exception as e: + msg = f"Invalid URL format: {e}" + raise ValueError(msg) from e + + try: + # ============================================================================ + # Step 3: Validate URL scheme (only http/https allowed) + # ============================================================================ + _validate_url_scheme(parsed.scheme) + if parsed.scheme not in ("http", "https"): + # Non-HTTP schemes (ftp, file, etc.) are not subject to SSRF protection + return url, [] + + # ============================================================================ + # Step 4: Extract and validate hostname + # ============================================================================ + hostname = _validate_hostname_exists(parsed.hostname) + + # ============================================================================ + # Step 5: Check allowlist (early return for trusted hosts) + # ============================================================================ + # Allowlisted hosts bypass all SSRF checks and DNS pinning + # This is used for legitimate internal services like Ollama (localhost) + if is_host_allowed(hostname): + logger.debug(f"Hostname {hostname} is in allowlist, bypassing SSRF checks and DNS pinning") + return url, [] + + # ============================================================================ + # Step 6: Handle direct IP addresses + # ============================================================================ + # Check if the hostname is already an IP address (no DNS resolution needed) + try: + ip_obj = ipaddress.ip_address(hostname) + + # Check if this specific IP is in the allowlist + if is_host_allowed(hostname, str(ip_obj)): + logger.debug(f"IP {hostname} is in allowlist") + return url, [] + + # Check if IP is in blocked ranges (private IPs, localhost, etc.) + if is_ip_blocked(ip_obj): + msg = ( + f"Access to IP address {hostname} is blocked by SSRF protection. " + "To allow this IP, add it to LANGFLOW_SSRF_ALLOWED_HOSTS environment variable." + ) + raise SSRFProtectionError(msg) + # Direct IP is public and allowed - return it for DNS pinning + # (Even though it's already an IP, we return it for consistency) + logger.debug(f"Direct IP {hostname} validated, will use for DNS pinning") + return url, [hostname] # noqa: TRY300 + + except ValueError: + # Not an IP address, it's a hostname - continue to DNS resolution + pass + + # ============================================================================ + # Step 7: Resolve hostname to IP addresses + # ============================================================================ + # This is the critical step for DNS pinning - we resolve DNS here during + # validation, and the returned IPs will be used for the actual HTTP request + resolved_ips = resolve_hostname(hostname) + blocked_ips = [] + + # ============================================================================ + # Step 8: Validate all resolved IPs + # ============================================================================ + # Security: We must check ALL resolved IPs before making any decisions. + # A hostname might resolve to multiple IPs (e.g., [8.8.8.8, 192.168.1.1]). + # If we return early on the first allowlisted IP, we skip validation of + # remaining IPs, which could include blocked/internal addresses. + # + # Strategy: + # 1. Collect all allowlisted IPs and all blocked IPs + # 2. If ANY IP is blocked → block the entire hostname (security first) + # 3. If some IPs are allowlisted but others are not → use only allowlisted IPs for pinning + # 4. If all IPs are public (none blocked, none allowlisted) → use all for pinning + allowed_ips = [] + for ip in resolved_ips: + # Check if this resolved IP is in the allowlist + if is_host_allowed(hostname, ip): + allowed_ips.append(ip) + # Check if IP is in blocked ranges + elif is_ip_blocked(ip): + blocked_ips.append(ip) + + # ============================================================================ + # Step 9: Block if any resolved IPs are private/internal + # ============================================================================ + # Security: If ANY resolved IP is blocked, we block the entire hostname. + # This prevents attacks where a hostname resolves to both safe and unsafe IPs. + if blocked_ips: + msg = ( + f"Hostname {hostname} resolves to blocked IP address(es): {', '.join(blocked_ips)}. " + "To allow this hostname, add it to LANGFLOW_SSRF_ALLOWED_HOSTS environment variable." + ) + raise SSRFProtectionError(msg) + + # ============================================================================ + # Step 9b: Handle partially allowlisted IPs + # ============================================================================ + # If some (but not all) IPs are allowlisted, use only the allowlisted ones for pinning + if allowed_ips: + logger.debug( + f"Hostname {hostname} has {len(allowed_ips)} allowlisted IP(s) out of {len(resolved_ips)} total. " + f"Using allowlisted IPs for DNS pinning: {allowed_ips}" + ) + return url, allowed_ips + # ============================================================================ + # Step 10: Return validated IPs for DNS pinning + # ============================================================================ + # All IPs are public and safe - return them for DNS pinning + # The HTTP client will use these IPs directly, preventing DNS rebinding + logger.debug(f"Hostname {hostname} validated, resolved to {resolved_ips}, will use for DNS pinning") + return url, resolved_ips # noqa: TRY300 + + except SSRFProtectionError: + # Re-raise SSRF errors as-is + raise + except Exception as e: + # Wrap unexpected errors in SSRFProtectionError + msg = f"Error validating URL: {e}" + raise SSRFProtectionError(msg) from e diff --git a/src/lfx/src/lfx/utils/ssrf_transport.py b/src/lfx/src/lfx/utils/ssrf_transport.py new file mode 100644 index 0000000000..86e6cdf20a --- /dev/null +++ b/src/lfx/src/lfx/utils/ssrf_transport.py @@ -0,0 +1,242 @@ +"""Custom httpx transport with DNS pinning for SSRF protection. + +This module provides a custom httpx transport that pins DNS resolution to validated +IP addresses, preventing DNS rebinding attacks that could bypass SSRF protection. + +The implementation uses a custom AsyncNetworkBackend to intercept TCP connections +and connect to the pinned IP address while preserving the original hostname for +TLS SNI (Server Name Indication) and certificate verification. +""" + +import ssl +from collections.abc import Iterable + +import httpcore +import httpx +from httpx import URL, Proxy +from httpx._config import create_ssl_context + +from lfx.logging import logger + + +class DNSPinningNetworkBackend(httpcore.AsyncNetworkBackend): + """Network backend that pins DNS resolution to validated IP addresses. + + This backend intercepts TCP connection attempts and redirects them to pinned + IP addresses while preserving the original hostname for TLS SNI and certificate + verification. This prevents DNS rebinding attacks without breaking HTTPS. + + How it works: + 1. When httpcore tries to connect to a hostname, we intercept the connect_tcp() call + 2. If the hostname has a pinned IP, we connect to that IP instead + 3. The original hostname is preserved for TLS handshake (SNI) and cert verification + 4. This prevents DNS rebinding while maintaining full HTTPS compatibility + """ + + def __init__(self, pinned_ips: dict[str, list[str]], backend: httpcore.AsyncNetworkBackend | None = None): + """Initialize the DNS pinning backend. + + Args: + pinned_ips: Dictionary mapping hostnames to list of validated IP addresses + backend: Underlying network backend (defaults to AnyIOBackend for asyncio) + """ + self.pinned_ips = pinned_ips + # Use httpcore's default async backend (AnyIOBackend) if none provided + # This is the public API recommended in httpcore documentation + if backend is None: + backend = httpcore.AnyIOBackend() + self._backend = backend + logger.debug(f"Created DNS pinning network backend with pinned IPs: {pinned_ips}") + + async def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + """Connect to TCP socket, using pinned IP if available. + + This method intercepts connection attempts and redirects to pinned IPs + while preserving the original hostname for TLS. + + Args: + host: Hostname to connect to (may be replaced with pinned IP) + port: Port number + timeout: Connection timeout + local_address: Local address to bind to + socket_options: Socket options + + Returns: + Network stream for the connection + """ + # Check if this hostname has pinned IPs + if host in self.pinned_ips: + pinned_ips = self.pinned_ips[host] + + # Security: If host is in pinned_ips but list is empty, fail rather than bypass + if not pinned_ips: + msg = f"DNS pinning: Host {host} is marked for pinning but has no pinned IPs" + logger.error(msg) + raise RuntimeError(msg) + + logger.debug(f"DNS pinning: Connecting to pinned IPs {pinned_ips} for hostname {host}") + + # Try each pinned IP in order (supports dual-stack and load balancing) + # The TLS layer will still use the original hostname for SNI and cert verification + last_error = None + for pinned_ip in pinned_ips: + try: + logger.debug(f"DNS pinning: Attempting connection to {pinned_ip}") + return await self._backend.connect_tcp( + host=pinned_ip, + port=port, + timeout=timeout, + local_address=local_address, + socket_options=socket_options, + ) + except (OSError, TimeoutError) as e: + last_error = e + logger.debug(f"DNS pinning: Failed to connect to {pinned_ip}: {e}") + continue + + # All pinned IPs failed, raise the last error + # This should never be None since we checked for empty list above + if last_error is None: + msg = f"DNS pinning: All pinned IPs failed for {host} but no error was captured" + raise RuntimeError(msg) + raise last_error + + # No pinned IP, use normal connection + return await self._backend.connect_tcp( + host=host, + port=port, + timeout=timeout, + local_address=local_address, + socket_options=socket_options, + ) + + async def connect_unix_socket( + self, + path: str, + timeout: float | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + """Connect to Unix socket (pass through to underlying backend).""" + return await self._backend.connect_unix_socket( + path=path, + timeout=timeout, + socket_options=socket_options, + ) + + async def sleep(self, seconds: float) -> None: + """Sleep for specified duration (pass through to underlying backend).""" + await self._backend.sleep(seconds) + + +class SSRFProtectedTransport(httpx.AsyncHTTPTransport): + """HTTP transport that pins DNS resolution to validated IPs. + + This transport prevents DNS rebinding attacks by using a custom network backend + that connects to pinned IP addresses while preserving the original hostname for + TLS SNI and certificate verification. + + Unlike the naive approach of rewriting URLs (which breaks HTTPS), this implementation + works at the network layer to ensure both security and compatibility. + + Example: + >>> pinned_ips = {"example.com": "93.184.216.34"} + >>> transport = SSRFProtectedTransport(pinned_ips=pinned_ips) + >>> async with httpx.AsyncClient(transport=transport) as client: + ... # Request to example.com will connect to 93.184.216.34 + ... # But TLS will still verify against example.com certificate + ... response = await client.get("https://example.com/path") + """ + + def __init__( + self, + pinned_ips: dict[str, list[str]], + verify: bool | str | ssl.SSLContext = True, # noqa: FBT001, FBT002 + cert: tuple[str, str] | tuple[str, str, str] | str | None = None, + trust_env: bool = True, # noqa: FBT001, FBT002 + http1: bool = True, # noqa: FBT001, FBT002 + http2: bool = False, # noqa: FBT001, FBT002 + limits: httpx.Limits = httpx.Limits(), # noqa: B008 + proxy: httpx._types.ProxyTypes | None = None, + uds: str | None = None, + local_address: str | None = None, + retries: int = 0, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ): + """Initialize transport with pinned DNS mappings. + + Args: + pinned_ips: Dictionary mapping hostnames to list of validated IP addresses. + Example: {"example.com": ["93.184.216.34", "2606:2800:220:1:248:1893:25c8:1946"]} + verify: SSL verification settings + cert: Client certificate + trust_env: Whether to trust environment variables for proxy config + http1: Enable HTTP/1.1 + http2: Enable HTTP/2 + limits: Connection pool limits + proxy: Proxy configuration + uds: Unix domain socket path + local_address: Local address to bind to + retries: Number of retries + socket_options: Socket options + """ + # Create custom network backend with DNS pinning + network_backend = DNSPinningNetworkBackend(pinned_ips=pinned_ips) + + # Create SSL context (same as parent class) + ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env) + + # Handle proxy (same as parent class) + if proxy is not None: + proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy + + # Create pool with our custom network backend + # We replicate the parent's logic but add network_backend parameter + if proxy is None: + self._pool = httpcore.AsyncConnectionPool( + ssl_context=ssl_context, + max_connections=limits.max_connections, + max_keepalive_connections=limits.max_keepalive_connections, + keepalive_expiry=limits.keepalive_expiry, + http1=http1, + http2=http2, + uds=uds, + local_address=local_address, + retries=retries, + socket_options=socket_options, + network_backend=network_backend, # Our custom backend! + ) + else: + # For proxy scenarios, we'd need to handle HTTPProxy/SOCKSProxy + # For now, raise an error as DNS pinning with proxies needs special handling + msg = "DNS pinning with proxies is not currently supported" + raise NotImplementedError(msg) + + self.pinned_ips = pinned_ips + logger.debug(f"Created SSRF protected transport with pinned IPs: {pinned_ips}") + + +def create_ssrf_protected_client( + hostname: str, validated_ips: list[str] | tuple[str, ...], **client_kwargs +) -> httpx.AsyncClient: + """Create an httpx client with DNS pinning for SSRF protection. + + Args: + hostname: The hostname to pin + validated_ips: List of validated IP addresses to use for this hostname. + IPs will be tried in order for dual-stack/load-balanced hosts. + **client_kwargs: Additional arguments for AsyncClient (e.g., timeout, headers) + + Returns: + Configured AsyncClient with DNS pinning + """ + # Convert to list if tuple + ip_list = list(validated_ips) if isinstance(validated_ips, tuple) else validated_ips + transport = SSRFProtectedTransport(pinned_ips={hostname: ip_list}) + return httpx.AsyncClient(transport=transport, **client_kwargs)