Draft: feat: add network isolation controls for high-compliance deployments

src/backend/base/langflow/services/store/factory.py

@@ -5,7 +5,7 @@
 from typing_extensions import override
 
 from langflow.services.factory import ServiceFactory
-from langflow.services.store.service import StoreService
+from langflow.services.store.service import DisabledStoreService, StoreService
 
 if TYPE_CHECKING:
     from lfx.services.settings.service import SettingsService
@@ -17,4 +17,6 @@ def __init__(self) -> None:
 
     @override
     def create(self, settings_service: SettingsService):
+        if not settings_service.settings.store:
+            return DisabledStoreService(settings_service)
         return StoreService(settings_service)

src/backend/base/langflow/services/store/service.py

@@ -72,6 +72,98 @@ def get_id_from_search_string(search_string: str) -> str | None:
     return possible_id
 
 
+class DisabledStoreService(Service):
+    """No-op implementation of StoreService for when store access is disabled."""
+
+    name = "store_service"
+
+    def __init__(self, settings_service: SettingsService):
+        self.settings_service = settings_service
+
+    async def check_api_key(self, _api_key: str) -> bool:
+        return False
+
+    async def get(
+        self, _url: str, _api_key: str | None = None, _params: dict[str, Any] | None = None
+    ) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+        return [], {}
+
+    async def call_webhook(self, _api_key: str, _webhook_url: str, _component_id: UUID) -> None:
+        return None
+
+    async def count_components(
+        self,
+        _filter_conditions: list[dict[str, Any]],
+        *,
+        _api_key: str | None = None,
+        _use_api_key: bool | None = False,
+    ) -> int:
+        return 0
+
+    async def query_components(
+        self,
+        *,
+        _api_key: str | None = None,
+        _sort: list[str] | None = None,
+        _page: int = 1,
+        _limit: int = 15,
+        _fields: list[str] | None = None,
+        _filter_conditions: list[dict[str, Any]] | None = None,
+        _use_api_key: bool | None = False,
+    ) -> tuple[list[ListComponentResponse], dict[str, Any]]:
+        return [], {}
+
+    async def get_liked_by_user_components(self, _component_ids: list[str], _api_key: str) -> list[str]:
+        return []
+
+    async def get_components_in_users_collection(self, _component_ids: list[str], _api_key: str) -> list[str]:
+        return []
+
+    async def download(self, _api_key: str, _component_id: UUID) -> DownloadComponentResponse:
+        msg = "Store access is disabled"
+        raise ValueError(msg)
+
+    async def upload(self, _api_key: str, _component_data: StoreComponentCreate) -> CreateComponentResponse:
+        msg = "Store access is disabled"
+        raise ValueError(msg)
+
+    async def update(
+        self, _api_key: str, _component_id: UUID, _component_data: StoreComponentCreate
+    ) -> CreateComponentResponse:
+        msg = "Store access is disabled"
+        raise ValueError(msg)
+
+    async def get_tags(self) -> list[dict[str, Any]]:
+        return []
+
+    async def get_user_likes(self, _api_key: str) -> list[dict[str, Any]]:
+        return []
+
+    async def get_component_likes_count(self, _component_id: str, _api_key: str | None = None) -> int:
+        return 0
+
+    async def like_component(self, _api_key: str, _component_id: str) -> bool:
+        return False
+
+    async def get_list_component_response_model(
+        self,
+        *,
+        _component_id: str | None = None,
+        _search: str | None = None,
+        _private: bool | None = None,
+        _tags: list[str] | None = None,
+        _is_component: bool | None = None,
+        _fields: list[str] | None = None,
+        _filter_by_user: bool = False,
+        _liked: bool = False,
+        _store_api_key: str | None = None,
+        _sort: list[str] | None = None,
+        _page: int = 1,
+        _limit: int = 15,
+    ) -> ListComponentResponseModel:
+        return ListComponentResponseModel(results=[], authorized=False, count=0)
+
+
 class StoreService(Service):
     """This is a service that integrates langflow with the store which is a Directus instance.
 

src/backend/base/langflow/utils/version.py

@@ -1,3 +1,4 @@
+import os
 from importlib import metadata
 
 import httpx
@@ -74,6 +75,10 @@ def is_nightly(v: str) -> bool:
 
 
 def fetch_latest_version(package_name: str, *, include_prerelease: bool) -> str | None:
+    # Skip version check if explicitly disabled via environment variable
+    if os.getenv("LANGFLOW_DISABLE_VERSION_CHECK", "false").lower() == "true":
+        return None
+
     package_name = package_name.replace(" ", "-").lower()
     try:
         response = httpx.get(f"https://pypi.org/pypi/{package_name}/json")

src/lfx/src/lfx/components/nvidia/nvidia.py

@@ -1,3 +1,4 @@
+import os
 from typing import Any
 
 from lfx.base.models.model import LCModelComponent
@@ -20,7 +21,12 @@ class NVIDIAModelComponent(LCModelComponent):
         warnings.filterwarnings("ignore", category=UserWarning, module="langchain_nvidia_ai_endpoints._common")
         from langchain_nvidia_ai_endpoints import ChatNVIDIA
 
-        all_models = ChatNVIDIA().get_available_models()
+        # Skip NVIDIA model fetching if explicitly disabled
+        if os.getenv("LANGFLOW_SKIP_NVIDIA_FETCH", "false").lower() == "true":
+            logger.info("NVIDIA model fetching disabled via LANGFLOW_SKIP_NVIDIA_FETCH environment variable.")
+            all_models = []
+        else:
+            all_models = ChatNVIDIA().get_available_models()
     except ImportError as e:
         msg = "Please install langchain-nvidia-ai-endpoints to use the NVIDIA model."
         raise ImportError(msg) from e

src/lfx/src/lfx/interface/components.py

@@ -89,6 +89,14 @@ def _read_component_index(custom_path: str | None = None) -> dict | None:
         if custom_path:
             # Check if it's a URL
             if custom_path.startswith(("http://", "https://")):
+                # Block remote component indices unless explicitly allowed
+                if os.getenv("LANGFLOW_ALLOW_REMOTE_COMPONENT_INDEX", "false").lower() != "true":
+                    logger.warning(
+                        "Remote component indices are disabled for security. "
+                        "Set LANGFLOW_ALLOW_REMOTE_COMPONENT_INDEX=true to allow."
+                    )
+                    return None
+
                 # Fetch from URL
                 import httpx