refactor(security): dedupe the local-file-access restriction reader

ssrf_protection had its own _local_file_access_restricted() that was byte-identical
to file_path_security.is_local_file_access_restricted() (same settings read + fail-open).
Reuse the public helper and drop the private duplicate; update the shared test mock to
patch both settings-read namespaces.
This commit is contained in:
Jordan Frazier
2026-06-23 08:44:14 -04:00
parent 2c35cfef1a
commit 0929fd93bb
2 changed files with 14 additions and 21 deletions

View File

@ -26,6 +26,7 @@ from urllib.parse import urlparse
from lfx.logging import logger
from lfx.services.deps import get_settings_service
from lfx.utils.file_path_security import is_local_file_access_restricted
class SSRFProtectionError(ValueError):
@ -463,18 +464,6 @@ def validate_connector_url_for_ssrf(url: str) -> None:
_LOCAL_FILE_DB_DIALECTS = frozenset({"sqlite", "duckdb", "access", "shell"})
def _local_file_access_restricted() -> bool:
"""Return True if LANGFLOW_RESTRICT_LOCAL_FILE_ACCESS is enabled."""
try:
return bool(get_settings_service().settings.restrict_local_file_access)
except Exception: # noqa: BLE001 - settings may be unavailable; default to not restricted
logger.warning(
"Could not read restrict_local_file_access setting; treating local file restriction "
"as DISABLED (fail-open to default). Local-file DB dialects are not being blocked."
)
return False
def validate_database_url_for_ssrf(url: str, *, validate_network_host: bool = True) -> None:
"""Validate a SQLAlchemy database URL against SSRF and local-file access.
@ -500,7 +489,7 @@ def validate_database_url_for_ssrf(url: str, *, validate_network_host: bool = Tr
ValueError: If the URL is malformed.
"""
ssrf_on = is_ssrf_protection_enabled() and validate_network_host
file_restricted = _local_file_access_restricted()
file_restricted = is_local_file_access_restricted()
if not ssrf_on and not file_restricted:
return
@ -610,7 +599,7 @@ def validate_git_repository_url(url: str) -> None:
pre_colon = url.split(":", 1)[0]
is_local_path = scheme == "file" or (scheme == "" and ("/" in pre_colon or url.startswith(("/", ".", "~"))))
if is_local_path:
if is_ssrf_protection_enabled() or _local_file_access_restricted():
if is_ssrf_protection_enabled() or is_local_file_access_restricted():
msg = "Cloning local-filesystem Git repositories is not permitted."
raise SSRFProtectionError(msg)
return

View File

@ -25,13 +25,17 @@ def mock_ssrf_settings(*, enabled=False, allowed_hosts=None, restrict_files=Fals
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
# Explicit (not a truthy MagicMock) so DB local-file checks behave deterministically.
mock_settings.settings.restrict_local_file_access = restrict_files
mock_get_settings.return_value = mock_settings
mock_settings = MagicMock()
mock_settings.settings.ssrf_protection_enabled = enabled
mock_settings.settings.ssrf_allowed_hosts = allowed_hosts
# Explicit (not a truthy MagicMock) so DB local-file checks behave deterministically.
mock_settings.settings.restrict_local_file_access = restrict_files
# The local-file-restriction read lives in file_path_security (is_local_file_access_restricted,
# reused by the DB/git validators here), so patch its settings source too.
with (
patch("lfx.utils.ssrf_protection.get_settings_service", return_value=mock_settings),
patch("lfx.utils.file_path_security.get_settings_service", return_value=mock_settings),
):
yield