mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 08:23:43 +08:00
feat: add SSRF protection configuration for API Requests (#10544)
* Add SSRF protection configuration for API Requests * clean up, add env vars to settings * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * [autofix.ci] apply automated fixes (attempt 3/3) * move ip blocklist to lazy load, simplify functions * [autofix.ci] apply automated fixes * refactor: replace global variable with functools.cache for IP ranges * refactor: simplify IP blocking logic with comprehension * refactor: remove unused url parameter from validation helpers * refactor: restructure exception handling in IP validation * refactor: use parenthesized context managers in SSRF tests * refactor: apply parenthesized context manager syntax in tests * Update docs to reflect behavior when params are mixed * docs update * whitespace * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * [autofix.ci] apply automated fixes * starter projects * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
This commit is contained in:
Jordan Frazier
File diff suppressed because one or more lines are too long
@ -1,4 +1,6 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import aiofiles
|
||||
import aiofiles.os
|
||||
@ -28,7 +30,7 @@ class TestAPIRequestComponent(ComponentTestBaseWithoutClient):
|
||||
"headers": [{"key": "User-Agent", "value": "test-agent"}],
|
||||
"body": [],
|
||||
"timeout": 30,
|
||||
"follow_redirects": True,
|
||||
"follow_redirects": False, # Changed default for SSRF security
|
||||
"save_to_file": False,
|
||||
"include_httpx_metadata": False,
|
||||
"mode": "URL",
|
||||
@ -339,3 +341,225 @@ class TestAPIRequestComponent(ComponentTestBaseWithoutClient):
|
||||
assert is_binary
|
||||
assert file_path is not None
|
||||
assert file_path.suffix == ".bin"
|
||||
|
||||
|
||||
class TestAPIRequestSSRFProtection:
|
||||
"""Test SSRF protection in API Request component."""
|
||||
|
||||
@pytest.fixture
|
||||
def component_class(self):
|
||||
"""Return the component class to test."""
|
||||
return APIRequestComponent
|
||||
|
||||
@pytest.fixture
|
||||
def default_kwargs(self):
|
||||
"""Return the default kwargs for the component."""
|
||||
return {
|
||||
"url_input": "https://example.com/api/test",
|
||||
"method": "GET",
|
||||
"headers": [{"key": "User-Agent", "value": "test-agent"}],
|
||||
"body": [],
|
||||
"timeout": 30,
|
||||
"follow_redirects": False, # Changed default for SSRF security
|
||||
"save_to_file": False,
|
||||
"include_httpx_metadata": False,
|
||||
"mode": "URL",
|
||||
"curl_input": "",
|
||||
"query_params": {},
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
async def component(self, component_class, default_kwargs):
|
||||
"""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
|
||||
component.url_input = "http://127.0.0.1:8080"
|
||||
|
||||
with 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)
|
||||
|
||||
async def test_ssrf_protection_enabled_blocks_localhost(self, component):
|
||||
"""Test that SSRF protection blocks localhost when enabled."""
|
||||
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.api_request.validate_url_for_ssrf") as mock_validate,
|
||||
):
|
||||
from lfx.utils.ssrf_protection import SSRFProtectionError
|
||||
|
||||
# Make it raise in enforcement mode
|
||||
mock_validate.side_effect = SSRFProtectionError("Access to 127.0.0.1 blocked")
|
||||
|
||||
with 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",
|
||||
]
|
||||
|
||||
with patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}):
|
||||
for url in private_ips:
|
||||
component.url_input = url
|
||||
|
||||
with patch("lfx.components.data.api_request.validate_url_for_ssrf") as mock_validate:
|
||||
from lfx.utils.ssrf_protection import SSRFProtectionError
|
||||
|
||||
mock_validate.side_effect = SSRFProtectionError(f"Access to {url} blocked")
|
||||
|
||||
with pytest.raises(ValueError, match="SSRF Protection"):
|
||||
await component.make_api_request()
|
||||
|
||||
async def test_ssrf_protection_enabled_blocks_metadata_endpoint(self, component):
|
||||
"""Test that SSRF protection blocks cloud metadata endpoints when enabled."""
|
||||
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.api_request.validate_url_for_ssrf") as mock_validate,
|
||||
):
|
||||
from lfx.utils.ssrf_protection import SSRFProtectionError
|
||||
|
||||
mock_validate.side_effect = SSRFProtectionError("Access to 169.254.169.254 blocked")
|
||||
|
||||
with 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."""
|
||||
public_urls = [
|
||||
"https://api.openai.com/v1/chat/completions",
|
||||
"https://api.github.com/repos/langflow-ai/langflow",
|
||||
"https://www.google.com",
|
||||
]
|
||||
|
||||
with patch.dict(os.environ, {"LANGFLOW_SSRF_PROTECTION_ENABLED": "true"}):
|
||||
for url in public_urls:
|
||||
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)
|
||||
|
||||
async def test_ssrf_protection_allowlist_bypass(self, component):
|
||||
"""Test that allowlisted hosts 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"},
|
||||
),
|
||||
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)
|
||||
|
||||
async def test_ssrf_protection_allowlist_cidr(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"},
|
||||
),
|
||||
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"
|
||||
|
||||
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"}))
|
||||
|
||||
# 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
|
||||
|
||||
async def test_follow_redirects_security_warning(self, component):
|
||||
"""Test that enabling follow_redirects logs a security warning."""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
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:
|
||||
respx.get("https://example.com/api").mock(return_value=Response(200, json={"status": "ok"}))
|
||||
|
||||
result = await component.make_api_request()
|
||||
assert isinstance(result, Data)
|
||||
|
||||
# Verify log was called with security warning
|
||||
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
|
||||
|
||||
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):
|
||||
"""Test that URLs without protocol get normalized to https://."""
|
||||
# Test URL without protocol
|
||||
component.url_input = "example.com"
|
||||
|
||||
with 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
|
||||
component.url_input = "http://example.com"
|
||||
|
||||
with 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"
|
||||
|
||||
with 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"
|
||||
|
||||
File diff suppressed because one or more lines are too long
@ -27,6 +27,7 @@ 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
|
||||
|
||||
# Define fields for each mode
|
||||
MODE_FIELDS = {
|
||||
@ -145,8 +146,13 @@ class APIRequestComponent(Component):
|
||||
BoolInput(
|
||||
name="follow_redirects",
|
||||
display_name="Follow Redirects",
|
||||
value=True,
|
||||
info="Whether to follow http redirects.",
|
||||
value=False,
|
||||
info=(
|
||||
"Whether to follow HTTP redirects. "
|
||||
"WARNING: Enabling redirects may allow SSRF bypass attacks where a public URL "
|
||||
"redirects to internal resources. Only enable if you trust the target server. "
|
||||
"See OWASP SSRF Prevention Cheat Sheet for details."
|
||||
),
|
||||
advanced=True,
|
||||
),
|
||||
BoolInput(
|
||||
@ -424,6 +430,14 @@ class APIRequestComponent(Component):
|
||||
save_to_file = self.save_to_file
|
||||
include_httpx_metadata = self.include_httpx_metadata
|
||||
|
||||
# Security warning when redirects are enabled
|
||||
if follow_redirects:
|
||||
self.log(
|
||||
"Security Warning: HTTP redirects are enabled. This may allow SSRF bypass attacks "
|
||||
"where a public URL redirects to internal resources (e.g., cloud metadata endpoints). "
|
||||
"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
|
||||
@ -437,6 +451,15 @@ class APIRequestComponent(Component):
|
||||
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
|
||||
try:
|
||||
validate_url_for_ssrf(url, warn_only=True)
|
||||
except SSRFProtectionError as e:
|
||||
# This will only raise if SSRF protection is enabled and warn_only=False
|
||||
msg = f"SSRF Protection: {e}"
|
||||
raise ValueError(msg) from e
|
||||
|
||||
# Process query parameters
|
||||
if isinstance(self.query_params, str):
|
||||
query_params = dict(parse_qsl(self.query_params))
|
||||
|
||||
@ -317,6 +317,23 @@ class Settings(BaseSettings):
|
||||
update_starter_projects: bool = True
|
||||
"""If set to True, Langflow will update starter projects."""
|
||||
|
||||
# SSRF Protection
|
||||
ssrf_protection_enabled: bool = False
|
||||
"""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
|
||||
including internal services, private networks, and cloud metadata endpoints.
|
||||
Default is False for backward compatibility. In v2.0, this will be changed to True.
|
||||
|
||||
Note: When ssrf_protection_enabled is disabled, the ssrf_allowed_hosts setting is ignored and has no effect."""
|
||||
ssrf_allowed_hosts: list[str] = []
|
||||
"""Comma-separated list of hosts/IPs/CIDR ranges to allow despite SSRF protection.
|
||||
Examples: 'internal-api.company.local,192.168.1.0/24,10.0.0.5,*.dev.internal'
|
||||
Supports exact hostnames, wildcard domains (*.example.com), exact IPs, and CIDR ranges.
|
||||
|
||||
Note: This setting only takes effect when ssrf_protection_enabled is True.
|
||||
When protection is disabled, all hosts are allowed regardless of this setting."""
|
||||
|
||||
@field_validator("cors_origins", mode="before")
|
||||
@classmethod
|
||||
def validate_cors_origins(cls, value):
|
||||
|
||||
384
src/lfx/src/lfx/utils/ssrf_protection.py
Normal file
384
src/lfx/src/lfx/utils/ssrf_protection.py
Normal file
@ -0,0 +1,384 @@
|
||||
"""SSRF (Server-Side Request Forgery) protection utilities.
|
||||
|
||||
This module provides validation to prevent SSRF attacks by blocking requests to:
|
||||
- Private IP ranges (RFC 1918)
|
||||
- Loopback addresses
|
||||
- Cloud metadata endpoints (169.254.169.254)
|
||||
- Other internal/special-use addresses
|
||||
|
||||
IMPORTANT: HTTP Redirects
|
||||
According to OWASP SSRF Prevention Cheat Sheet, HTTP redirects should be DISABLED
|
||||
to prevent bypass attacks where a public URL redirects to internal resources.
|
||||
The API Request component has (as of v1.7.0) follow_redirects=False by default.
|
||||
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_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
|
||||
import ipaddress
|
||||
import socket
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from lfx.logging import logger
|
||||
from lfx.services.deps import get_settings_service
|
||||
|
||||
|
||||
class SSRFProtectionError(ValueError):
|
||||
"""Raised when a URL is blocked due to SSRF protection."""
|
||||
|
||||
|
||||
@functools.cache
|
||||
def get_blocked_ip_ranges() -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
|
||||
"""Get the list of blocked IP ranges, initializing lazily on first access.
|
||||
|
||||
This lazy loading avoids the startup cost of creating all ip_network objects
|
||||
at module import time.
|
||||
|
||||
Returns:
|
||||
list: List of blocked IPv4 and IPv6 network ranges.
|
||||
"""
|
||||
return [
|
||||
# IPv4 ranges
|
||||
ipaddress.ip_network("0.0.0.0/8"), # Current network (only valid as source)
|
||||
ipaddress.ip_network("10.0.0.0/8"), # Private network (RFC 1918)
|
||||
ipaddress.ip_network("100.64.0.0/10"), # Carrier-grade NAT (RFC 6598)
|
||||
ipaddress.ip_network("127.0.0.0/8"), # Loopback
|
||||
ipaddress.ip_network("169.254.0.0/16"), # Link-local / AWS metadata
|
||||
ipaddress.ip_network("172.16.0.0/12"), # Private network (RFC 1918)
|
||||
ipaddress.ip_network("192.0.0.0/24"), # IETF Protocol Assignments
|
||||
ipaddress.ip_network("192.0.2.0/24"), # Documentation (TEST-NET-1)
|
||||
ipaddress.ip_network("192.168.0.0/16"), # Private network (RFC 1918)
|
||||
ipaddress.ip_network("198.18.0.0/15"), # Benchmarking
|
||||
ipaddress.ip_network("198.51.100.0/24"), # Documentation (TEST-NET-2)
|
||||
ipaddress.ip_network("203.0.113.0/24"), # Documentation (TEST-NET-3)
|
||||
ipaddress.ip_network("224.0.0.0/4"), # Multicast
|
||||
ipaddress.ip_network("240.0.0.0/4"), # Reserved
|
||||
ipaddress.ip_network("255.255.255.255/32"), # Broadcast
|
||||
# IPv6 ranges
|
||||
ipaddress.ip_network("::1/128"), # Loopback
|
||||
ipaddress.ip_network("::/128"), # Unspecified address
|
||||
ipaddress.ip_network("::ffff:0:0/96"), # IPv4-mapped IPv6 addresses
|
||||
ipaddress.ip_network("100::/64"), # Discard prefix
|
||||
ipaddress.ip_network("2001::/23"), # IETF Protocol Assignments
|
||||
ipaddress.ip_network("2001:db8::/32"), # Documentation
|
||||
ipaddress.ip_network("fc00::/7"), # Unique local addresses (ULA)
|
||||
ipaddress.ip_network("fe80::/10"), # Link-local
|
||||
ipaddress.ip_network("ff00::/8"), # Multicast
|
||||
]
|
||||
|
||||
|
||||
def is_ssrf_protection_enabled() -> bool:
|
||||
"""Check if SSRF protection is enabled in settings.
|
||||
|
||||
Returns:
|
||||
bool: True if SSRF protection is enabled, False otherwise.
|
||||
"""
|
||||
return get_settings_service().settings.ssrf_protection_enabled
|
||||
|
||||
|
||||
def get_allowed_hosts() -> list[str]:
|
||||
"""Get list of allowed hosts and/or CIDR ranges for SSRF protection.
|
||||
|
||||
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()]
|
||||
|
||||
|
||||
def is_host_allowed(hostname: str, ip: str | None = None) -> bool:
|
||||
"""Check if a hostname or IP is in the allowed hosts list.
|
||||
|
||||
Args:
|
||||
hostname: Hostname to check
|
||||
ip: Optional IP address to check
|
||||
|
||||
Returns:
|
||||
bool: True if hostname or IP is in the allowed list, False otherwise.
|
||||
"""
|
||||
allowed_hosts = get_allowed_hosts()
|
||||
if not allowed_hosts:
|
||||
return False
|
||||
|
||||
# Check hostname match
|
||||
if hostname in allowed_hosts:
|
||||
return True
|
||||
|
||||
# Check if hostname matches any wildcard patterns
|
||||
for allowed in allowed_hosts:
|
||||
if allowed.startswith("*."):
|
||||
# Wildcard domain matching
|
||||
domain_suffix = allowed[1:] # Remove the *
|
||||
if hostname.endswith(domain_suffix) or hostname == domain_suffix[1:]:
|
||||
return True
|
||||
|
||||
# Check IP-based matching if IP is provided
|
||||
if ip:
|
||||
try:
|
||||
ip_obj = ipaddress.ip_address(ip)
|
||||
|
||||
# Check exact IP match
|
||||
if ip in allowed_hosts:
|
||||
return True
|
||||
|
||||
# Check CIDR range match
|
||||
for allowed in allowed_hosts:
|
||||
try:
|
||||
# Try to parse as CIDR network
|
||||
if "/" in allowed:
|
||||
network = ipaddress.ip_network(allowed, strict=False)
|
||||
if ip_obj in network:
|
||||
return True
|
||||
except (ValueError, ipaddress.AddressValueError):
|
||||
# Not a valid CIDR, skip
|
||||
continue
|
||||
|
||||
except (ValueError, ipaddress.AddressValueError):
|
||||
# Invalid IP, skip IP-based checks
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def is_ip_blocked(ip: str | ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
||||
"""Check if an IP address is in a blocked range.
|
||||
|
||||
Args:
|
||||
ip: IP address to check (string or ipaddress object)
|
||||
|
||||
Returns:
|
||||
bool: True if IP is in a blocked range, False otherwise.
|
||||
"""
|
||||
try:
|
||||
ip_obj = ipaddress.ip_address(ip) if isinstance(ip, str) else ip
|
||||
|
||||
# Check against all blocked ranges
|
||||
return any(ip_obj in blocked_range for blocked_range in get_blocked_ip_ranges())
|
||||
except (ValueError, ipaddress.AddressValueError):
|
||||
# If we can't parse the IP, treat it as blocked for safety
|
||||
return True
|
||||
|
||||
|
||||
def resolve_hostname(hostname: str) -> list[str]:
|
||||
"""Resolve a hostname to its IP addresses.
|
||||
|
||||
Args:
|
||||
hostname: Hostname to resolve
|
||||
|
||||
Returns:
|
||||
list[str]: List of resolved IP addresses
|
||||
|
||||
Raises:
|
||||
SSRFProtectionError: If hostname cannot be resolved
|
||||
"""
|
||||
try:
|
||||
# Get address info for both IPv4 and IPv6
|
||||
addr_info = socket.getaddrinfo(hostname, None)
|
||||
|
||||
# Extract unique IP addresses
|
||||
ips = []
|
||||
for info in addr_info:
|
||||
ip = info[4][0]
|
||||
# Remove IPv6 zone ID if present (e.g., "fe80::1%eth0" -> "fe80::1")
|
||||
if "%" in ip:
|
||||
ip = ip.split("%")[0]
|
||||
if ip not in ips:
|
||||
ips.append(ip)
|
||||
|
||||
if not ips:
|
||||
msg = f"Unable to resolve hostname: {hostname}"
|
||||
raise SSRFProtectionError(msg)
|
||||
except socket.gaierror as e:
|
||||
msg = f"DNS resolution failed for {hostname}: {e}"
|
||||
raise SSRFProtectionError(msg) from e
|
||||
except Exception as e:
|
||||
msg = f"Error resolving hostname {hostname}: {e}"
|
||||
raise SSRFProtectionError(msg) from e
|
||||
|
||||
return ips
|
||||
|
||||
|
||||
def _validate_url_scheme(scheme: str) -> None:
|
||||
"""Validate that URL scheme is http or https.
|
||||
|
||||
Args:
|
||||
scheme: URL scheme to validate
|
||||
|
||||
Raises:
|
||||
SSRFProtectionError: If scheme is invalid
|
||||
"""
|
||||
if scheme not in ("http", "https"):
|
||||
msg = f"Invalid URL scheme '{scheme}'. Only http and https are allowed."
|
||||
raise SSRFProtectionError(msg)
|
||||
|
||||
|
||||
def _validate_hostname_exists(hostname: str | None) -> str:
|
||||
"""Validate that hostname exists in the URL.
|
||||
|
||||
Args:
|
||||
hostname: Hostname to validate (may be None)
|
||||
|
||||
Returns:
|
||||
str: The validated hostname
|
||||
|
||||
Raises:
|
||||
SSRFProtectionError: If hostname is missing
|
||||
"""
|
||||
if not hostname:
|
||||
msg = "URL must contain a valid hostname"
|
||||
raise SSRFProtectionError(msg)
|
||||
return hostname
|
||||
|
||||
|
||||
def _validate_direct_ip_address(hostname: str) -> bool:
|
||||
"""Validate a direct IP address in the URL.
|
||||
|
||||
Args:
|
||||
hostname: Hostname that may be an IP address
|
||||
|
||||
Returns:
|
||||
bool: True if hostname is a direct IP and validation passed,
|
||||
False if hostname is not an IP (caller should continue with DNS resolution)
|
||||
|
||||
Raises:
|
||||
SSRFProtectionError: If IP is blocked
|
||||
"""
|
||||
try:
|
||||
ip_obj = ipaddress.ip_address(hostname)
|
||||
except ValueError:
|
||||
# Not an IP address, it's a hostname - caller should continue with DNS resolution
|
||||
return False
|
||||
|
||||
# It's a direct IP address
|
||||
# Check if IP is in allowlist
|
||||
if is_host_allowed(hostname, str(ip_obj)):
|
||||
logger.debug("IP address %s is in allowlist, bypassing SSRF checks", hostname)
|
||||
return True
|
||||
|
||||
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)
|
||||
|
||||
# Direct IP is allowed (public IP)
|
||||
return True
|
||||
|
||||
|
||||
def _validate_hostname_resolution(hostname: str) -> None:
|
||||
"""Resolve hostname and validate resolved IPs are not blocked.
|
||||
|
||||
Args:
|
||||
hostname: Hostname to resolve and validate
|
||||
|
||||
Raises:
|
||||
SSRFProtectionError: If resolved IPs are blocked
|
||||
"""
|
||||
# Resolve hostname to IP addresses
|
||||
try:
|
||||
resolved_ips = resolve_hostname(hostname)
|
||||
except SSRFProtectionError:
|
||||
# Re-raise SSRF errors as-is
|
||||
raise
|
||||
except Exception as e:
|
||||
msg = f"Failed to resolve hostname {hostname}: {e}"
|
||||
raise SSRFProtectionError(msg) from e
|
||||
|
||||
# Check if any resolved IP is blocked
|
||||
blocked_ips = []
|
||||
for ip in resolved_ips:
|
||||
# Check if this specific IP is in the allowlist
|
||||
if is_host_allowed(hostname, ip):
|
||||
logger.debug("Resolved IP %s for hostname %s is in allowlist, bypassing SSRF checks", ip, hostname)
|
||||
return
|
||||
|
||||
if is_ip_blocked(ip):
|
||||
blocked_ips.append(ip)
|
||||
|
||||
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:
|
||||
"""Validate a URL to prevent SSRF attacks.
|
||||
|
||||
This function performs the following checks:
|
||||
1. Validates the URL scheme (only http/https allowed)
|
||||
2. Validates hostname exists
|
||||
3. Checks if hostname/IP is in allowlist
|
||||
4. If direct IP: validates it's not in blocked ranges
|
||||
5. If hostname: resolves to IPs and validates they're not in blocked ranges
|
||||
|
||||
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)
|
||||
|
||||
Raises:
|
||||
SSRFProtectionError: If the URL is blocked due to SSRF protection (only if warn_only=False)
|
||||
ValueError: If the URL is malformed
|
||||
"""
|
||||
# Skip validation if SSRF protection is disabled
|
||||
if not is_ssrf_protection_enabled():
|
||||
return
|
||||
|
||||
# Parse URL
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
except Exception as e:
|
||||
msg = f"Invalid URL format: {e}"
|
||||
raise ValueError(msg) from e
|
||||
|
||||
try:
|
||||
# Validate scheme
|
||||
_validate_url_scheme(parsed.scheme)
|
||||
if parsed.scheme not in ("http", "https"):
|
||||
return
|
||||
|
||||
# Validate hostname exists
|
||||
hostname = _validate_hostname_exists(parsed.hostname)
|
||||
|
||||
# Check if hostname/IP is in allowlist (early return if allowed)
|
||||
if is_host_allowed(hostname):
|
||||
logger.debug("Hostname %s is in allowlist, bypassing SSRF checks", hostname)
|
||||
return
|
||||
|
||||
# Validate direct IP address or resolve hostname
|
||||
is_direct_ip = _validate_direct_ip_address(hostname)
|
||||
if is_direct_ip:
|
||||
# Direct IP was handled (allowed or exception raised)
|
||||
return
|
||||
|
||||
# Not a direct IP, resolve hostname and validate
|
||||
_validate_hostname_resolution(hostname)
|
||||
except SSRFProtectionError as e:
|
||||
if warn_only:
|
||||
logger.warning("SSRF Protection Warning: %s [URL: %s]", str(e), url)
|
||||
logger.warning(
|
||||
"This request will be blocked when SSRF protection is enforced in the next major version. "
|
||||
"Please review your API Request components."
|
||||
)
|
||||
return
|
||||
raise
|
||||
435
src/lfx/tests/unit/utils/test_ssrf_protection.py
Normal file
435
src/lfx/tests/unit/utils/test_ssrf_protection.py
Normal file
@ -0,0 +1,435 @@
|
||||
"""Unit tests for SSRF protection utilities."""
|
||||
|
||||
from contextlib import contextmanager
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from lfx.utils.ssrf_protection import (
|
||||
SSRFProtectionError,
|
||||
get_allowed_hosts,
|
||||
is_host_allowed,
|
||||
is_ip_blocked,
|
||||
is_ssrf_protection_enabled,
|
||||
resolve_hostname,
|
||||
validate_url_for_ssrf,
|
||||
)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def mock_ssrf_settings(*, enabled=False, allowed_hosts=None):
|
||||
"""Context manager to mock SSRF settings."""
|
||||
if allowed_hosts is None:
|
||||
allowed_hosts = []
|
||||
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_protection_enabled = enabled
|
||||
mock_settings.settings.ssrf_allowed_hosts = allowed_hosts
|
||||
mock_get_settings.return_value = mock_settings
|
||||
yield
|
||||
|
||||
|
||||
class TestSSRFProtectionConfiguration:
|
||||
"""Test SSRF protection configuration and environment variables."""
|
||||
|
||||
def test_ssrf_protection_disabled_by_default(self):
|
||||
"""Test that SSRF protection is disabled by default (for now)."""
|
||||
# TODO: Update this test when default changes to enabled in v2.0
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_protection_enabled = False
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_ssrf_protection_enabled() is False
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("setting_value", "expected"),
|
||||
[
|
||||
(True, True),
|
||||
(False, False),
|
||||
],
|
||||
)
|
||||
def test_ssrf_protection_setting(self, setting_value, expected):
|
||||
"""Test SSRF protection setting value."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_protection_enabled = setting_value
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_ssrf_protection_enabled() == expected
|
||||
|
||||
def test_allowed_hosts_empty_by_default(self):
|
||||
"""Test that allowed hosts is empty by default."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = []
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert get_allowed_hosts() == []
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("setting_value", "expected"),
|
||||
[
|
||||
([], []),
|
||||
(["example.com"], ["example.com"]),
|
||||
(["example.com", "api.example.com"], ["example.com", "api.example.com"]),
|
||||
(["192.168.1.0/24", "10.0.0.5"], ["192.168.1.0/24", "10.0.0.5"]),
|
||||
([" example.com ", " api.example.com "], ["example.com", "api.example.com"]),
|
||||
(["*.example.com"], ["*.example.com"]),
|
||||
(["", "example.com", " ", "api.example.com"], ["example.com", "api.example.com"]), # Test filtering
|
||||
],
|
||||
)
|
||||
def test_allowed_hosts_parsing(self, setting_value, expected):
|
||||
"""Test allowed hosts list cleaning and filtering."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = setting_value
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert get_allowed_hosts() == expected
|
||||
|
||||
|
||||
class TestIPBlocking:
|
||||
"""Test IP address blocking functionality."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ip",
|
||||
[
|
||||
# Loopback
|
||||
"127.0.0.1",
|
||||
"127.0.0.2",
|
||||
"127.255.255.255",
|
||||
"::1",
|
||||
# Private networks (RFC 1918)
|
||||
"10.0.0.1",
|
||||
"10.255.255.255",
|
||||
"172.16.0.1",
|
||||
"172.31.255.255",
|
||||
"192.168.0.1",
|
||||
"192.168.255.255",
|
||||
# Link-local / Cloud metadata
|
||||
"169.254.0.1",
|
||||
"169.254.169.254", # AWS/GCP/Azure metadata
|
||||
"169.254.255.255",
|
||||
# Carrier-grade NAT
|
||||
"100.64.0.1",
|
||||
"100.127.255.255",
|
||||
# Documentation/Test ranges
|
||||
"192.0.2.1",
|
||||
"198.51.100.1",
|
||||
"203.0.113.1",
|
||||
# Multicast
|
||||
"224.0.0.1",
|
||||
"239.255.255.255",
|
||||
# Reserved
|
||||
"240.0.0.1",
|
||||
"255.255.255.254",
|
||||
# Broadcast
|
||||
"255.255.255.255",
|
||||
# IPv6 ranges
|
||||
"fc00::1", # ULA
|
||||
"fe80::1", # Link-local
|
||||
"ff00::1", # Multicast
|
||||
],
|
||||
)
|
||||
def test_blocked_ips(self, ip):
|
||||
"""Test that private/internal IPs are blocked."""
|
||||
assert is_ip_blocked(ip) is True
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ip",
|
||||
[
|
||||
# Public IPv4 addresses
|
||||
"8.8.8.8", # Google DNS
|
||||
"1.1.1.1", # Cloudflare DNS
|
||||
"93.184.216.34", # example.com
|
||||
"151.101.1.140", # Reddit
|
||||
"13.107.42.14", # Microsoft
|
||||
# Public IPv6 addresses
|
||||
"2001:4860:4860::8888", # Google DNS
|
||||
"2606:4700:4700::1111", # Cloudflare DNS
|
||||
],
|
||||
)
|
||||
def test_allowed_ips(self, ip):
|
||||
"""Test that public IPs are allowed."""
|
||||
assert is_ip_blocked(ip) is False
|
||||
|
||||
def test_invalid_ip_is_blocked(self):
|
||||
"""Test that invalid IPs are treated as blocked for safety."""
|
||||
assert is_ip_blocked("not.an.ip.address") is True
|
||||
assert is_ip_blocked("999.999.999.999") is True
|
||||
|
||||
|
||||
class TestHostnameAllowlist:
|
||||
"""Test hostname allowlist functionality."""
|
||||
|
||||
def test_exact_hostname_match(self):
|
||||
"""Test exact hostname matching in allowlist."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = ["internal.company.local"]
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_host_allowed("internal.company.local") is True
|
||||
assert is_host_allowed("other.company.local") is False
|
||||
|
||||
def test_wildcard_hostname_match(self):
|
||||
"""Test wildcard hostname matching in allowlist."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = ["*.company.local"]
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_host_allowed("api.company.local") is True
|
||||
assert is_host_allowed("internal.company.local") is True
|
||||
assert is_host_allowed("company.local") is True
|
||||
assert is_host_allowed("other.domain.com") is False
|
||||
|
||||
def test_exact_ip_match(self):
|
||||
"""Test exact IP matching in allowlist."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = ["192.168.1.5"]
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_host_allowed("example.com", "192.168.1.5") is True
|
||||
assert is_host_allowed("example.com", "192.168.1.6") is False
|
||||
|
||||
def test_cidr_range_match(self):
|
||||
"""Test CIDR range matching in allowlist."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = ["192.168.1.0/24", "10.0.0.0/16"]
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_host_allowed("example.com", "192.168.1.5") is True
|
||||
assert is_host_allowed("example.com", "192.168.1.255") is True
|
||||
assert is_host_allowed("example.com", "192.168.2.5") is False
|
||||
assert is_host_allowed("example.com", "10.0.0.1") is True
|
||||
assert is_host_allowed("example.com", "10.0.255.255") is True
|
||||
assert is_host_allowed("example.com", "10.1.0.1") is False
|
||||
|
||||
def test_multiple_allowed_hosts(self):
|
||||
"""Test multiple entries in allowlist."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = ["internal.local", "192.168.1.0/24", "*.api.company.com"]
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_host_allowed("internal.local") is True
|
||||
assert is_host_allowed("v1.api.company.com") is True
|
||||
assert is_host_allowed("example.com", "192.168.1.100") is True
|
||||
assert is_host_allowed("other.com", "10.0.0.1") is False
|
||||
|
||||
def test_empty_allowlist(self):
|
||||
"""Test that empty allowlist returns False."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_allowed_hosts = []
|
||||
mock_get_settings.return_value = mock_settings
|
||||
assert is_host_allowed("example.com") is False
|
||||
assert is_host_allowed("example.com", "192.168.1.1") is False
|
||||
|
||||
|
||||
class TestHostnameResolution:
|
||||
"""Test DNS hostname resolution."""
|
||||
|
||||
def test_resolve_public_hostname(self):
|
||||
"""Test resolving a public hostname."""
|
||||
# Use a stable public hostname
|
||||
ips = resolve_hostname("dns.google")
|
||||
assert len(ips) > 0
|
||||
# Should resolve to public IPs (8.8.8.8 or 8.8.4.4)
|
||||
assert any(not is_ip_blocked(ip) for ip in ips)
|
||||
|
||||
def test_resolve_localhost(self):
|
||||
"""Test resolving localhost."""
|
||||
ips = resolve_hostname("localhost")
|
||||
assert len(ips) > 0
|
||||
# Should include 127.0.0.1 or ::1
|
||||
assert any(ip in ("127.0.0.1", "::1") for ip in ips)
|
||||
|
||||
def test_resolve_invalid_hostname(self):
|
||||
"""Test that invalid hostnames raise SSRFProtectionError."""
|
||||
with pytest.raises(SSRFProtectionError, match="DNS resolution failed"):
|
||||
resolve_hostname("this-hostname-definitely-does-not-exist-12345.invalid")
|
||||
|
||||
|
||||
class TestURLValidation:
|
||||
"""Test URL validation for SSRF protection."""
|
||||
|
||||
def test_protection_disabled_allows_all(self):
|
||||
"""Test that when protection is disabled, all URLs are allowed."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_protection_enabled = False
|
||||
mock_get_settings.return_value = mock_settings
|
||||
# These should all pass without errors when protection is disabled
|
||||
validate_url_for_ssrf("http://127.0.0.1", warn_only=False)
|
||||
validate_url_for_ssrf("http://169.254.169.254", warn_only=False)
|
||||
validate_url_for_ssrf("http://192.168.1.1", warn_only=False)
|
||||
|
||||
def test_invalid_scheme_blocked(self):
|
||||
"""Test that non-http/https schemes are blocked."""
|
||||
with patch("lfx.utils.ssrf_protection.get_settings_service") as mock_get_settings:
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.ssrf_protection_enabled = True
|
||||
mock_get_settings.return_value = mock_settings
|
||||
|
||||
with pytest.raises(SSRFProtectionError, match="Invalid URL scheme"):
|
||||
validate_url_for_ssrf("ftp://example.com", warn_only=False)
|
||||
|
||||
with pytest.raises(SSRFProtectionError, match="Invalid URL scheme"):
|
||||
validate_url_for_ssrf("file:///etc/passwd", warn_only=False)
|
||||
|
||||
def test_valid_schemes_allowed(self):
|
||||
"""Test that http and https schemes are explicitly allowed."""
|
||||
with (
|
||||
mock_ssrf_settings(enabled=True),
|
||||
patch("lfx.utils.ssrf_protection.resolve_hostname") as mock_resolve,
|
||||
):
|
||||
mock_resolve.return_value = ["93.184.216.34"] # Public IP (example.com)
|
||||
|
||||
# Should not raise - valid schemes with public IPs
|
||||
validate_url_for_ssrf("http://example.com", warn_only=False)
|
||||
validate_url_for_ssrf("https://example.com", warn_only=False)
|
||||
validate_url_for_ssrf("https://api.example.com/v1", warn_only=False)
|
||||
|
||||
def test_direct_ip_blocking(self):
|
||||
"""Test blocking of direct IP addresses."""
|
||||
with mock_ssrf_settings(enabled=True):
|
||||
# Loopback
|
||||
with pytest.raises(SSRFProtectionError, match="blocked"):
|
||||
validate_url_for_ssrf("http://127.0.0.1", warn_only=False)
|
||||
|
||||
# Private network
|
||||
with pytest.raises(SSRFProtectionError, match="blocked"):
|
||||
validate_url_for_ssrf("http://192.168.1.1", warn_only=False)
|
||||
|
||||
# Metadata endpoint
|
||||
with pytest.raises(SSRFProtectionError, match="blocked"):
|
||||
validate_url_for_ssrf("http://169.254.169.254/latest/meta-data/", warn_only=False)
|
||||
|
||||
def test_public_ips_allowed(self):
|
||||
"""Test that public IP addresses are allowed."""
|
||||
with mock_ssrf_settings(enabled=True):
|
||||
# Should not raise
|
||||
validate_url_for_ssrf("http://8.8.8.8", warn_only=False)
|
||||
validate_url_for_ssrf("http://1.1.1.1", warn_only=False)
|
||||
|
||||
def test_public_hostnames_allowed(self):
|
||||
"""Test that public hostnames are allowed."""
|
||||
with mock_ssrf_settings(enabled=True):
|
||||
# Test with real DNS to stable Google service
|
||||
validate_url_for_ssrf("https://www.google.com", warn_only=False)
|
||||
|
||||
# Mock DNS for other domains
|
||||
with patch("lfx.utils.ssrf_protection.resolve_hostname") as mock_resolve:
|
||||
mock_resolve.return_value = ["93.184.216.34"] # Public IP
|
||||
validate_url_for_ssrf("https://api.example.com", warn_only=False)
|
||||
validate_url_for_ssrf("https://example.com", warn_only=False)
|
||||
|
||||
def test_localhost_hostname_blocked(self):
|
||||
"""Test that localhost hostname is blocked."""
|
||||
with mock_ssrf_settings(enabled=True), pytest.raises(SSRFProtectionError, match="blocked IP address"):
|
||||
validate_url_for_ssrf("http://localhost:8080", warn_only=False)
|
||||
|
||||
def test_allowlist_bypass_hostname(self):
|
||||
"""Test that allowlisted hostnames bypass SSRF checks."""
|
||||
with mock_ssrf_settings(enabled=True, allowed_hosts=["internal.company.local"]):
|
||||
# Should not raise even if it resolves to private IP
|
||||
# (We can't easily test actual resolution without mocking, but the allowlist check happens first)
|
||||
validate_url_for_ssrf("http://internal.company.local", warn_only=False)
|
||||
|
||||
def test_allowlist_bypass_ip(self):
|
||||
"""Test that allowlisted IPs bypass SSRF checks."""
|
||||
with mock_ssrf_settings(enabled=True, allowed_hosts=["192.168.1.5"]):
|
||||
# Should not raise
|
||||
validate_url_for_ssrf("http://192.168.1.5", warn_only=False)
|
||||
|
||||
def test_allowlist_bypass_cidr(self):
|
||||
"""Test that IPs in allowlisted CIDR ranges bypass SSRF checks."""
|
||||
with mock_ssrf_settings(enabled=True, allowed_hosts=["192.168.1.0/24"]):
|
||||
# Should not raise
|
||||
validate_url_for_ssrf("http://192.168.1.5", warn_only=False)
|
||||
validate_url_for_ssrf("http://192.168.1.100", warn_only=False)
|
||||
|
||||
def test_warn_only_mode_logs_warnings(self):
|
||||
"""Test that warn_only mode logs warnings instead of raising errors."""
|
||||
with mock_ssrf_settings(enabled=True), patch("lfx.utils.ssrf_protection.logger") as mock_logger:
|
||||
# Should not raise, but should log warning
|
||||
validate_url_for_ssrf("http://127.0.0.1", warn_only=True)
|
||||
|
||||
# Check that warning was logged
|
||||
mock_logger.warning.assert_called()
|
||||
assert any("SSRF Protection Warning" in str(call) for call in mock_logger.warning.call_args_list)
|
||||
|
||||
def test_malformed_url_raises_value_error(self):
|
||||
"""Test that malformed URLs raise ValueError."""
|
||||
with mock_ssrf_settings(enabled=True), pytest.raises(ValueError, match="Invalid URL"):
|
||||
validate_url_for_ssrf("not a valid url", warn_only=False)
|
||||
|
||||
def test_missing_hostname_blocked(self):
|
||||
"""Test that URLs without hostname are blocked."""
|
||||
with mock_ssrf_settings(enabled=True), pytest.raises(SSRFProtectionError, match="valid hostname"):
|
||||
validate_url_for_ssrf("http://", warn_only=False)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"http://[::1]", # IPv6 loopback
|
||||
"http://[::1]:8080/admin",
|
||||
"http://[fc00::1]", # IPv6 ULA
|
||||
"http://[fe80::1]", # IPv6 link-local
|
||||
],
|
||||
)
|
||||
def test_ipv6_blocking(self, url):
|
||||
"""Test that private IPv6 addresses are blocked."""
|
||||
with mock_ssrf_settings(enabled=True), pytest.raises(SSRFProtectionError, match="blocked"):
|
||||
validate_url_for_ssrf(url, warn_only=False)
|
||||
|
||||
def test_ipv6_public_allowed(self):
|
||||
"""Test that public IPv6 addresses are allowed."""
|
||||
with mock_ssrf_settings(enabled=True):
|
||||
# Should not raise
|
||||
validate_url_for_ssrf("http://[2001:4860:4860::8888]", warn_only=False)
|
||||
|
||||
|
||||
class TestIntegrationScenarios:
|
||||
"""Test realistic integration scenarios."""
|
||||
|
||||
def test_aws_metadata_blocked(self):
|
||||
"""Test that AWS metadata endpoint is blocked."""
|
||||
with mock_ssrf_settings(enabled=True), pytest.raises(SSRFProtectionError):
|
||||
validate_url_for_ssrf("http://169.254.169.254/latest/meta-data/iam/security-credentials/", warn_only=False)
|
||||
|
||||
def test_internal_admin_panel_blocked(self):
|
||||
"""Test that internal admin panels are blocked."""
|
||||
with mock_ssrf_settings(enabled=True), pytest.raises(SSRFProtectionError):
|
||||
validate_url_for_ssrf("http://192.168.1.1/admin", warn_only=False)
|
||||
|
||||
def test_legitimate_api_allowed(self):
|
||||
"""Test that legitimate external APIs are allowed."""
|
||||
with (
|
||||
mock_ssrf_settings(enabled=True),
|
||||
patch("lfx.utils.ssrf_protection.resolve_hostname") as mock_resolve,
|
||||
):
|
||||
mock_resolve.return_value = ["104.16.132.229"] # Public IP
|
||||
|
||||
# Should all pass - mocked as public IPs
|
||||
validate_url_for_ssrf("https://api.openai.com/v1/chat/completions", warn_only=False)
|
||||
validate_url_for_ssrf("https://api.github.com/repos/langflow-ai/langflow", warn_only=False)
|
||||
validate_url_for_ssrf("https://www.googleapis.com/auth/cloud-platform", warn_only=False)
|
||||
|
||||
def test_docker_internal_networking_requires_allowlist(self):
|
||||
"""Test that Docker internal networking requires allowlist configuration."""
|
||||
with (
|
||||
mock_ssrf_settings(enabled=True),
|
||||
patch("lfx.utils.ssrf_protection.resolve_hostname") as mock_resolve,
|
||||
):
|
||||
mock_resolve.return_value = ["172.18.0.2"] # Docker bridge network IP
|
||||
|
||||
# Without allowlist, should be blocked
|
||||
with pytest.raises(SSRFProtectionError):
|
||||
validate_url_for_ssrf("http://database:5432", warn_only=False)
|
||||
|
||||
# With allowlist, should be allowed
|
||||
with (
|
||||
mock_ssrf_settings(enabled=True, allowed_hosts=["database", "*.internal.local"]),
|
||||
patch("lfx.utils.ssrf_protection.resolve_hostname") as mock_resolve,
|
||||
):
|
||||
mock_resolve.return_value = ["172.18.0.2"] # Docker bridge network IP
|
||||
|
||||
validate_url_for_ssrf("http://database:5432", warn_only=False)
|
||||
validate_url_for_ssrf("http://api.internal.local", warn_only=False)
|
||||
Reference in New Issue
Block a user