Langflow SSRF 服务器端请求伪造漏洞(CVE-2025-68477)
Langflow SSRF 服务器端请求伪造漏洞(CVE-2025-68477)
什么是LangFlow？
Langflow是一个功能强大的平台，用于构建和部署 AI 驱动的代理和工作流。它为开发者提供可视化创作体验以及内置的 API 和 MCP 服务器，可以将每个工作流转化为可集成到基于任何框架或技术栈构建的应用程序中的工具。Langflow 功能齐全，支持所有主流的 LLM、向量数据库以及不断增长的 AI 工具库。
环境搭建
本文使用LangFlow 1.6.9版本
创建python虚拟环境
{
"version": "0.2.0",
"configurations": [
{
"name": "Python 调试程序: 模块",
"type": "debugpy",
"request": "launch",
"module": "langflow",
"args": ["run", "--frontend-path", "${workspaceFolder}/src/frontend/build"],
"python": "${workspaceFolder}/.venv/Scripts/python.exe"
}
]
}首先执行uv run langflow run --backend-only，uv会自动安装相关的依赖
然后运行pnpm install，pnpm run build，再进行调试
组件
根据官方文档，在LangFlow中组件是流程的构建基石。就像应用程序中的类一样，每个组件都针对特定的用例或集成而设计。
可以看见核心组件包括，Input/Output，Data Source-API Request，Mock Data，URL，Web Search
API Request组件
根据漏洞名字为SSRF漏洞，应该与HTTP请求有关，在API Request组件中使用 URL 或 curl 命令构建和发送 HTTP 请求：
https://docs.langflow.org/api-request
本地新建一个API-REQUEST组件
点击code可以查看此组件的代码实现
漏洞代码位置：src/backend/base/langflow/components/data/api_request.py
关键漏洞点分析：
- URL 规范化方法缺少 SSRF 防护（第281-290行）：
def _normalize_url(self, url: str) -> str:
"""Normalize URL by adding https:// if no protocol is specified."""
if not url or not isinstance(url, str):
msg = "URL cannot be empty"
raise ValueError(msg)
url = url.strip()
if url.startswith(("http://", "https://")):
return url
return f"https://{url}"问题： 该方法只检查URL是否包含协议，如果没有协议就添加 https://，但完全没有检查URL是否为内网地址（如 127.0.0.1、localhost、内网IP等）。
- URL 验证只检查格式，不检查内网地址（第417-439行）：
async def make_api_request(self) -> Data:
"""Make HTTP request with optimized parameter handling."""
method = self.method
url = self.url_input.strip() if isinstance(self.url_input, str) else ""
headers = self.headers or {}
body = self.body or {}
timeout = self.timeout
follow_redirects = self.follow_redirects
save_to_file = self.save_to_file
include_httpx_metadata = self.include_httpx_metadata
# Normalize URL before validation
url = self._normalize_url(url)
# Validate URL
if not validators.url(url):
msg = f"Invalid URL provided: {url}"
raise ValueError(msg)问题： 使用 validators.url() 只验证URL格式是否正确，不验证是否为内网地址，允许攻击者传入 http://127.0.0.1:8000 等内网地址。
- 直接发送HTTP请求，无任何SSRF防护（第452-465行）：
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问题： 直接使用 httpx.AsyncClient 发送请求，没有任何SSRF防护机制，可以访问任意内网地址。
完整组件代码：
import json
import re
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
import aiofiles
import aiofiles.os as aiofiles_os
import httpx
import validators
from langflow.base.curl.parse import parse_context
from langflow.custom.custom_component.component import Component
from langflow.inputs.inputs import TabInput
from langflow.io import (
BoolInput,
DataInput,
DropdownInput,
IntInput,
MessageTextInput,
MultilineInput,
Output,
TableInput,
)
from langflow.schema.data import Data
from langflow.schema.dotdict import dotdict
from langflow.services.deps import get_settings_service
from langflow.utils.component_utils import set_current_fields, set_field_advanced, set_field_display
# Define fields for each mode
MODE_FIELDS = {
"URL": [
"url_input",
"method",
],
"cURL": ["curl_input"],
}
# Fields that should always be visible
DEFAULT_FIELDS = ["mode"]
class APIRequestComponent(Component):
display_name = "API Request"
description = "Make HTTP requests using URL or cURL commands."
documentation: str = "https://docs.langflow.org/components-data#api-request"
icon = "Globe"
name = "APIRequest"
inputs = [
MessageTextInput(
name="url_input",
display_name="URL",
info="Enter the URL for the request.",
advanced=False,
tool_mode=True,
),
MultilineInput(
name="curl_input",
display_name="cURL",
info=(
"Paste a curl command to populate the fields. "
"This will fill in the dictionary fields for headers and body."
),
real_time_refresh=True,
tool_mode=True,
advanced=True,
show=False,
),
DropdownInput(
name="method",
display_name="Method",
options=["GET", "POST", "PATCH", "PUT", "DELETE"],
value="GET",
info="The HTTP method to use.",
real_time_refresh=True,
),
TabInput(
name="mode",
display_name="Mode",
options=["URL", "cURL"],
value="URL",
info="Enable cURL mode to populate fields from a cURL command.",
real_time_refresh=True,
),
DataInput(
name="query_params",
display_name="Query Parameters",
info="The query parameters to append to the URL.",
advanced=True,
),
TableInput(
name="body",
display_name="Body",
info="The body to send with the request as a dictionary (for POST, PATCH, PUT).",
table_schema=[
{
"name": "key",
"display_name": "Key",
"type": "str",
"description": "Parameter name",
},
{
"name": "value",
"display_name": "Value",
"description": "Parameter value",
},
],
value=[],
input_types=["Data"],
advanced=True,
real_time_refresh=True,
),
TableInput(
name="headers",
display_name="Headers",
info="The headers to send with the request",
table_schema=[
{
"name": "key",
"display_name": "Header",
"type": "str",
"description": "Header name",
},
{
"name": "value",
"display_name": "Value",
"type": "str",
"description": "Header value",
},
],
value=[{"key": "User-Agent", "value": get_settings_service().settings.user_agent}],
advanced=True,
input_types=["Data"],
real_time_refresh=True,
),
IntInput(
name="timeout",
display_name="Timeout",
value=30,
info="The timeout to use for the request.",
advanced=True,
),
BoolInput(
name="follow_redirects",
display_name="Follow Redirects",
value=True,
info="Whether to follow http redirects.",
advanced=True,
),
BoolInput(
name="save_to_file",
display_name="Save to File",
value=False,
info="Save the API response to a temporary file",
advanced=True,
),
BoolInput(
name="include_httpx_metadata",
display_name="Include HTTPx Metadata",
value=False,
info=(
"Include properties such as headers, status_code, response_headers, "
"and redirection_history in the output."
),
advanced=True,
),
]
outputs = [
Output(display_name="API Response", name="data", method="make_api_request"),
]
def _parse_json_value(self, value: Any) -> Any:
"""Parse a value that might be a JSON string."""
if not isinstance(value, str):
return value
try:
parsed = json.loads(value)
except json.JSONDecodeError:
return value
else:
return parsed
def _process_body(self, body: Any) -> dict:
"""Process the body input into a valid dictionary."""
if body is None:
return {}
if hasattr(body, "data"):
body = body.data
if isinstance(body, dict):
return self._process_dict_body(body)
if isinstance(body, str):
return self._process_string_body(body)
if isinstance(body, list):
return self._process_list_body(body)
return {}
def _process_dict_body(self, body: dict) -> dict:
"""Process dictionary body by parsing JSON values."""
return {k: self._parse_json_value(v) for k, v in body.items()}
def _process_string_body(self, body: str) -> dict:
"""Process string body by attempting JSON parse."""
try:
return self._process_body(json.loads(body))
except json.JSONDecodeError:
return {"data": body}
def _process_list_body(self, body: list) -> dict:
"""Process list body by converting to key-value dictionary."""
processed_dict = {}
try:
for item in body:
# Unwrap Data objects
current_item = item
if hasattr(item, "data"):
unwrapped_data = item.data
# If the unwrapped data is a dict but not key-value format, use it directly
if isinstance(unwrapped_data, dict) and not self._is_valid_key_value_item(unwrapped_data):
return unwrapped_data
current_item = unwrapped_data
if not self._is_valid_key_value_item(current_item):
continue
key = current_item["key"]
value = self._parse_json_value(current_item["value"])
processed_dict[key] = value
except (KeyError, TypeError, ValueError) as e:
self.log(f"Failed to process body list: {e}")
return {}
return processed_dict
def _is_valid_key_value_item(self, item: Any) -> bool:
"""Check if an item is a valid key-value dictionary."""
return isinstance(item, dict) and "key" in item and "value" in item
def parse_curl(self, curl: str, build_config: dotdict) -> dotdict:
"""Parse a cURL command and update build configuration."""
try:
parsed = parse_context(curl)
# Update basic configuration
url = parsed.url
# Normalize URL before setting it
url = self._normalize_url(url)
build_config["url_input"]["value"] = url
build_config["method"]["value"] = parsed.method.upper()
# Process headers
headers_list = [{"key": k, "value": v} for k, v in parsed.headers.items()]
build_config["headers"]["value"] = headers_list
# Process body data
if not parsed.data:
build_config["body"]["value"] = []
elif parsed.data:
try:
json_data = json.loads(parsed.data)
if isinstance(json_data, dict):
body_list = [
{"key": k, "value": json.dumps(v) if isinstance(v, dict | list) else str(v)}
for k, v in json_data.items()
]
build_config["body"]["value"] = body_list
else:
build_config["body"]["value"] = [{"key": "data", "value": json.dumps(json_data)}]
except json.JSONDecodeError:
build_config["body"]["value"] = [{"key": "data", "value": parsed.data}]
except Exception as exc:
msg = f"Error parsing curl: {exc}"
self.log(msg)
raise ValueError(msg) from exc
return build_config
def _normalize_url(self, url: str) -> str:
"""Normalize URL by adding https:// if no protocol is specified."""
if not url or not isinstance(url, str):
msg = "URL cannot be empty"
raise ValueError(msg)
url = url.strip()
if url.startswith(("http://", "https://")):
return url
return f"https://{url}"
async def make_request(
self,
client: httpx.AsyncClient,
method: str,
url: str,
headers: dict | None = None,
body: Any = None,
timeout: int = 5,
*,
follow_redirects: bool = True,
save_to_file: bool = False,
include_httpx_metadata: bool = False,
) -> Data:
method = method.upper()
if method not in {"GET", "POST", "PATCH", "PUT", "DELETE"}:
msg = f"Unsupported method: {method}"
raise ValueError(msg)
processed_body = self._process_body(body)
redirection_history = []
try:
# Prepare request parameters
request_params = {
"method": method,
"url": url,
"headers": headers,
"json": processed_body,
"timeout": timeout,
"follow_redirects": follow_redirects,
}
response = await client.request(**request_params)
redirection_history = [
{
"url": redirect.headers.get("Location", str(redirect.url)),
"status_code": redirect.status_code,
}
for redirect in response.history
]
is_binary, file_path = await self._response_info(response, with_file_path=save_to_file)
response_headers = self._headers_to_dict(response.headers)
# Base metadata
metadata = {
"source": url,
"status_code": response.status_code,
"response_headers": response_headers,
}
if redirection_history:
metadata["redirection_history"] = redirection_history
if save_to_file:
mode = "wb" if is_binary else "w"
encoding = response.encoding if mode == "w" else None
if file_path:
await aiofiles_os.makedirs(file_path.parent, exist_ok=True)
if is_binary:
async with aiofiles.open(file_path, "wb") as f:
await f.write(response.content)
await f.flush()
else:
async with aiofiles.open(file_path, "w", encoding=encoding) as f:
await f.write(response.text)
await f.flush()
metadata["file_path"] = str(file_path)
if include_httpx_metadata:
metadata.update({"headers": headers})
return Data(data=metadata)
# Handle response content
if is_binary:
result = response.content
else:
try:
result = response.json()
except json.JSONDecodeError:
self.log("Failed to decode JSON response")
result = response.text.encode("utf-8")
metadata["result"] = result
if include_httpx_metadata:
metadata.update({"headers": headers})
return Data(data=metadata)
except (httpx.HTTPError, httpx.RequestError, httpx.TimeoutException) as exc:
self.log(f"Error making request to {url}")
return Data(
data={
"source": url,
"headers": headers,
"status_code": 500,
"error": str(exc),
**({"redirection_history": redirection_history} if redirection_history else {}),
},
)
def add_query_params(self, url: str, params: dict) -> str:
"""Add query parameters to URL efficiently."""
if not params:
return url
url_parts = list(urlparse(url))
query = dict(parse_qsl(url_parts[4]))
query.update(params)
url_parts[4] = urlencode(query)
return urlunparse(url_parts)
def _headers_to_dict(self, headers: httpx.Headers) -> dict[str, str]:
"""Convert HTTP headers to a dictionary with lowercased keys."""
return {k.lower(): v for k, v in headers.items()}
def _process_headers(self, headers: Any) -> dict:
"""Process the headers input into a valid dictionary."""
if headers is None:
return {}
if isinstance(headers, dict):
return headers
if isinstance(headers, list):
return {item["key"]: item["value"] for item in headers if self._is_valid_key_value_item(item)}
return {}
async def make_api_request(self) -> Data:
"""Make HTTP request with optimized parameter handling."""
method = self.method
url = self.url_input.strip() if isinstance(self.url_input, str) else ""
headers = self.headers or {}
body = self.body or {}
timeout = self.timeout
follow_redirects = self.follow_redirects
save_to_file = self.save_to_file
include_httpx_metadata = self.include_httpx_metadata
# 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
url = self._normalize_url(url)
# Validate URL
if not validators.url(url):
msg = f"Invalid URL provided: {url}"
raise ValueError(msg)
# Process query parameters
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
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,
)
self.status = result
return result
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:
"""Update the build config based on the selected mode."""
if field_name != "mode":
if field_name == "curl_input" and self.mode == "cURL" and self.curl_input:
return self.parse_curl(self.curl_input, build_config)
return build_config
# print(f"Current mode: {field_value}")
if field_value == "cURL":
set_field_display(build_config, "curl_input", value=True)
if build_config["curl_input"]["value"]:
build_config = self.parse_curl(build_config["curl_input"]["value"], build_config)
else:
set_field_display(build_config, "curl_input", value=False)
return set_current_fields(
build_config=build_config,
action_fields=MODE_FIELDS,
selected_action=field_value,
default_fields=DEFAULT_FIELDS,
func=set_field_advanced,
default_value=True,
)
async def _response_info(
self, response: httpx.Response, *, with_file_path: bool = False
) -> tuple[bool, Path | None]:
"""Determine the file path and whether the response content is binary.
Args:
response (Response): The HTTP response object.
with_file_path (bool): Whether to save the response content to a file.
Returns:
Tuple[bool, Path | None]:
A tuple containing a boolean indicating if the content is binary and the full file path (if applicable).
"""
content_type = response.headers.get("Content-Type", "")
is_binary = "application/octet-stream" in content_type or "application/binary" in content_type
if not with_file_path:
return is_binary, None
component_temp_dir = Path(tempfile.gettempdir()) / self.__class__.__name__
# Create directory asynchronously
await aiofiles_os.makedirs(component_temp_dir, exist_ok=True)
filename = None
if "Content-Disposition" in response.headers:
content_disposition = response.headers["Content-Disposition"]
filename_match = re.search(r'filename="(.+?)"', content_disposition)
if filename_match:
extracted_filename = filename_match.group(1)
filename = extracted_filename
# Step 3: Infer file extension or use part of the request URL if no filename
if not filename:
# Extract the last segment of the URL path
url_path = urlparse(str(response.request.url) if response.request else "").path
base_name = Path(url_path).name # Get the last segment of the path
if not base_name: # If the path ends with a slash or is empty
base_name = "response"
# Infer file extension
content_type_to_extension = {
"text/plain": ".txt",
"application/json": ".json",
"image/jpeg": ".jpg",
"image/png": ".png",
"application/octet-stream": ".bin",
}
extension = content_type_to_extension.get(content_type, ".bin" if is_binary else ".txt")
filename = f"{base_name}{extension}"
# Step 4: Define the full file path
file_path = component_temp_dir / filename
# Step 5: Check if file exists asynchronously and handle accordingly
try:
# Try to create the file exclusively (x mode) to check existence
async with aiofiles.open(file_path, "x") as _:
pass # File created successfully, we can use this path
except FileExistsError:
# If file exists, append a timestamp to the filename
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S%f")
file_path = component_temp_dir / f"{timestamp}-{filename}"
return is_binary, file_path使用虚拟机进行测试一下，访问物理机的127.0.0.1:8000端口，成功访问。
URL 组件
URL组件同样也可以触发SSRF请求，访问物理机的8000端口
漏洞代码位置：src/backend/base/langflow/components/data/url.py
关键漏洞点分析：
- URL 规范化方法缺少 SSRF 防护（第184-204行）：
def ensure_url(self, url: str) -> str:
"""Ensures the given string is a valid URL.
Args:
url: The URL string to validate and normalize
Returns:
str: The normalized URL
Raises:
ValueError: If the URL is invalid
"""
url = url.strip()
if not url.startswith(("http://", "https://")):
url = "https://" + url
if not self.validate_url(url):
msg = f"Invalid URL: {url}"
raise ValueError(msg)
return url问题： 该方法只检查URL格式（通过正则表达式），不检查是否为内网地址。
- URL 验证正则表达式允许内网地址（第172-182行）：
@staticmethod
def validate_url(url: str) -> bool:
"""Validates if the given string matches URL pattern.
Args:
url: The URL string to validate
Returns:
bool: True if the URL is valid, False otherwise
"""
return bool(URL_REGEX.match(url))问题：URL_REGEX 正则表达式（第20-23行）只验证URL格式，不限制内网地址：
URL_REGEX = re.compile(
r"^(https?:\/\/)?" r"(www\.)?" r"([a-zA-Z0-9.-]+)" r"(\.[a-zA-Z]{2,})?" r"(:\d+)?" r"(\/[^\s]*)?$",
re.IGNORECASE,
)该正则允许 127.0.0.1、localhost、内网IP等。
- 直接使用 RecursiveUrlLoader 加载URL，无SSRF防护（第206-232行）：
def _create_loader(self, url: str) -> RecursiveUrlLoader:
"""Creates a RecursiveUrlLoader instance with the configured settings.
Args:
url: The URL to load
Returns:
RecursiveUrlLoader: Configured loader instance
"""
headers_dict = {header["key"]: header["value"] for header in self.headers if header["value"] is not None}
extractor = (lambda x: x) if self.format == "HTML" else (lambda x: BeautifulSoup(x, "lxml").get_text())
return RecursiveUrlLoader(
url=url,
max_depth=self.max_depth,
prevent_outside=self.prevent_outside,
use_async=self.use_async,
extractor=extractor,
timeout=self.timeout,
headers=headers_dict,
check_response_status=self.check_response_status,
continue_on_failure=self.continue_on_failure,
base_url=url, # Add base_url to ensure consistent domain crawling
autoset_encoding=self.autoset_encoding, # Enable automatic encoding detection
exclude_dirs=[], # Allow customization of excluded directories
link_regex=None, # Allow customization of link filtering
)问题： 虽然 prevent_outside 参数可以限制跨域访问，但对于初始URL本身没有SSRF防护，攻击者可以直接传入内网地址。
完整组件代码：
import re
import requests
from bs4 import BeautifulSoup
from langchain_community.document_loaders import RecursiveUrlLoader
from langflow.custom.custom_component.component import Component
from langflow.field_typing.range_spec import RangeSpec
from langflow.helpers.data import safe_convert
from langflow.io import BoolInput, DropdownInput, IntInput, MessageTextInput, Output, SliderInput, TableInput
from langflow.logging.logger import logger
from langflow.schema.dataframe import DataFrame
from langflow.schema.message import Message
from langflow.services.deps import get_settings_service
# Constants
DEFAULT_TIMEOUT = 30
DEFAULT_MAX_DEPTH = 1
DEFAULT_FORMAT = "Text"
URL_REGEX = re.compile(
r"^(https?:\/\/)?" r"(www\.)?" r"([a-zA-Z0-9.-]+)" r"(\.[a-zA-Z]{2,})?" r"(:\d+)?" r"(\/[^\s]*)?$",
re.IGNORECASE,
)
class URLComponent(Component):
"""A component that loads and parses content from web pages recursively.
This component allows fetching content from one or more URLs, with options to:
- Control crawl depth
- Prevent crawling outside the root domain
- Use async loading for better performance
- Extract either raw HTML or clean text
- Configure request headers and timeouts
"""
display_name = "URL"
description = "Fetch content from one or more web pages, following links recursively."
documentation: str = "https://docs.langflow.org/components-data#url"
icon = "layout-template"
name = "URLComponent"
inputs = [
MessageTextInput(
name="urls",
display_name="URLs",
info="Enter one or more URLs to crawl recursively, by clicking the '+' button.",
is_list=True,
tool_mode=True,
placeholder="Enter a URL...",
list_add_label="Add URL",
input_types=[],
),
SliderInput(
name="max_depth",
display_name="Depth",
info=(
"Controls how many 'clicks' away from the initial page the crawler will go:\n"
"- depth 1: only the initial page\n"
"- depth 2: initial page + all pages linked directly from it\n"
"- depth 3: initial page + direct links + links found on those direct link pages\n"
"Note: This is about link traversal, not URL path depth."
),
value=DEFAULT_MAX_DEPTH,
range_spec=RangeSpec(min=1, max=5, step=1),
required=False,
min_label=" ",
max_label=" ",
min_label_icon="None",
max_label_icon="None",
# slider_input=True
),
BoolInput(
name="prevent_outside",
display_name="Prevent Outside",
info=(
"If enabled, only crawls URLs within the same domain as the root URL. "
"This helps prevent the crawler from going to external websites."
),
value=True,
required=False,
advanced=True,
),
BoolInput(
name="use_async",
display_name="Use Async",
info=(
"If enabled, uses asynchronous loading which can be significantly faster "
"but might use more system resources."
),
value=True,
required=False,
advanced=True,
),
DropdownInput(
name="format",
display_name="Output Format",
info="Output Format. Use 'Text' to extract the text from the HTML or 'HTML' for the raw HTML content.",
options=["Text", "HTML"],
value=DEFAULT_FORMAT,
advanced=True,
),
IntInput(
name="timeout",
display_name="Timeout",
info="Timeout for the request in seconds.",
value=DEFAULT_TIMEOUT,
required=False,
advanced=True,
),
TableInput(
name="headers",
display_name="Headers",
info="The headers to send with the request",
table_schema=[
{
"name": "key",
"display_name": "Header",
"type": "str",
"description": "Header name",
},
{
"name": "value",
"display_name": "Value",
"type": "str",
"description": "Header value",
},
],
value=[{"key": "User-Agent", "value": get_settings_service().settings.user_agent}],
advanced=True,
input_types=["DataFrame"],
),
BoolInput(
name="filter_text_html",
display_name="Filter Text/HTML",
info="If enabled, filters out text/css content type from the results.",
value=True,
required=False,
advanced=True,
),
BoolInput(
name="continue_on_failure",
display_name="Continue on Failure",
info="If enabled, continues crawling even if some requests fail.",
value=True,
required=False,
advanced=True,
),
BoolInput(
name="check_response_status",
display_name="Check Response Status",
info="If enabled, checks the response status of the request.",
value=False,
required=False,
advanced=True,
),
BoolInput(
name="autoset_encoding",
display_name="Autoset Encoding",
info="If enabled, automatically sets the encoding of the request.",
value=True,
required=False,
advanced=True,
),
]
outputs = [
Output(display_name="Extracted Pages", name="page_results", method="fetch_content"),
Output(display_name="Raw Content", name="raw_results", method="fetch_content_as_message", tool_mode=False),
]
@staticmethod
def validate_url(url: str) -> bool:
"""Validates if the given string matches URL pattern.
Args:
url: The URL string to validate
Returns:
bool: True if the URL is valid, False otherwise
"""
return bool(URL_REGEX.match(url))
def ensure_url(self, url: str) -> str:
"""Ensures the given string is a valid URL.
Args:
url: The URL string to validate and normalize
Returns:
str: The normalized URL
Raises:
ValueError: If the URL is invalid
"""
url = url.strip()
if not url.startswith(("http://", "https://")):
url = "https://" + url
if not self.validate_url(url):
msg = f"Invalid URL: {url}"
raise ValueError(msg)
return url
def _create_loader(self, url: str) -> RecursiveUrlLoader:
"""Creates a RecursiveUrlLoader instance with the configured settings.
Args:
url: The URL to load
Returns:
RecursiveUrlLoader: Configured loader instance
"""
headers_dict = {header["key"]: header["value"] for header in self.headers if header["value"] is not None}
extractor = (lambda x: x) if self.format == "HTML" else (lambda x: BeautifulSoup(x, "lxml").get_text())
return RecursiveUrlLoader(
url=url,
max_depth=self.max_depth,
prevent_outside=self.prevent_outside,
use_async=self.use_async,
extractor=extractor,
timeout=self.timeout,
headers=headers_dict,
check_response_status=self.check_response_status,
continue_on_failure=self.continue_on_failure,
base_url=url, # Add base_url to ensure consistent domain crawling
autoset_encoding=self.autoset_encoding, # Enable automatic encoding detection
exclude_dirs=[], # Allow customization of excluded directories
link_regex=None, # Allow customization of link filtering
)
def fetch_url_contents(self) -> list[dict]:
"""Load documents from the configured URLs.
Returns:
List[Data]: List of Data objects containing the fetched content
Raises:
ValueError: If no valid URLs are provided or if there's an error loading documents
"""
try:
urls = list({self.ensure_url(url) for url in self.urls if url.strip()})
logger.debug(f"URLs: {urls}")
if not urls:
msg = "No valid URLs provided."
raise ValueError(msg)
all_docs = []
for url in urls:
logger.debug(f"Loading documents from {url}")
try:
loader = self._create_loader(url)
docs = loader.load()
if not docs:
logger.warning(f"No documents found for {url}")
continue
logger.debug(f"Found {len(docs)} documents from {url}")
all_docs.extend(docs)
except requests.exceptions.RequestException as e:
logger.exception(f"Error loading documents from {url}: {e}")
continue
if not all_docs:
msg = "No documents were successfully loaded from any URL"
raise ValueError(msg)
# data = [Data(text=doc.page_content, **doc.metadata) for doc in all_docs]
data = [
{
"text": safe_convert(doc.page_content, clean_data=True),
"url": doc.metadata.get("source", ""),
"title": doc.metadata.get("title", ""),
"description": doc.metadata.get("description", ""),
"content_type": doc.metadata.get("content_type", ""),
"language": doc.metadata.get("language", ""),
}
for doc in all_docs
]
except Exception as e:
error_msg = e.message if hasattr(e, "message") else e
msg = f"Error loading documents: {error_msg!s}"
logger.exception(msg)
raise ValueError(msg) from e
return data
def fetch_content(self) -> DataFrame:
"""Convert the documents to a DataFrame."""
return DataFrame(data=self.fetch_url_contents())
def fetch_content_as_message(self) -> Message:
"""Convert the documents to a Message."""
url_contents = self.fetch_url_contents()
return Message(text="\n\n".join([x["text"] for x in url_contents]), data={"data": url_contents})使用api请求
POST /api/v1/run/69bb68ec-4092-4f2b-b962-626a643f6fd2 HTTP/1.1
Host: 192.168.2.4:7860
User-Agent: python-requests/2.31.0
Accept-Encoding: gzip, deflate, br
Accept: */*
Connection: keep-alive
x-api-key: sk--ZwJxLCzu9JpFmCoKLB2OcGxh045qQbGjDeZd981srM
Content-Length: 163
Content-Type: application/json
{"output_type": "chat", "input_type": "chat", "input_value": "get file from http://127.0.0.1:8000/test.txt?", "session_id": "c937b766-c454-4c35-bfb9-eae14424cd1b"}漏洞共性分析
两个组件都存在相同的SSRF漏洞，根本原因如下：
1. 缺少内网地址检测
两个组件都只验证URL格式，不检查是否为内网地址：
- API Request组件：使用 validators.url() 只验证格式
- URL组件：使用正则表达式 URL_REGEX 只验证格式
应该被阻止的内网地址包括：
- 127.0.0.1 / localhost / 0.0.0.0
- 私有IP地址段：
- 10.0.0.0/8
- 172.16.0.0/12
- 192.168.0.0/16
- IPv6 内网地址：::1 / [::1]
- 域名解析到内网IP的情况
2. 缺少URL白名单机制
两个组件都允许访问任意URL，没有白名单限制。
3. 重定向未受控
API Request组件的 follow_redirects 参数默认为 True，攻击者可以通过重定向绕过某些防护。即使有URL验证，攻击者也可以先访问一个合法的外网URL，然后通过HTTP重定向到内网地址。
4. 修复建议
- 添加内网地址检测：检查URL的hostname是否为内网地址
- 在URL验证时添加检查：在 _normalize_url 或 ensure_url 方法中添加内网地址检查
- 限制重定向：设置 follow_redirects=False 或限制重定向次数，并检查重定向目标
- 使用URL白名单（如果适用）：只允许访问预定义的白名单域名/IP
5. 影响范围
- 版本：Langflow 1.6.9 及之前版本
- 组件：API Request 组件、URL 组件
- 攻击场景：
- 访问内网服务（如 Redis、MySQL、内网API等）
- 端口扫描内网主机
- 读取内网文件（如果内网服务支持）
- 绕过防火墙限制
6. 验证方法
使用以下Payload验证漏洞：
API Request组件：
http://127.0.0.1:8000/test.txt
http://localhost:3306
http://192.168.1.1:8080URL组件：
127.0.0.1:8000/test.txt
localhost:3306
192.168.1.1:8080通过API调用：
{
"output_type": "chat",
"input_type": "chat",
"input_value": "get file from http://127.0.0.1:8000/test.txt?",
"session_id": "xxx"
}END
AI 摘要AI 生成
LangFlow 1.6.9中API Request和URL组件缺乏内网地址检测，导致SSRF漏洞，可访问127.0.0.1、localhost及私有IP段。
由 AI 根据全文自动生成，仅供参考，以原文为准