feat(catalog): filter component palette by policy

src/backend/base/langflow/api/v1/chat.py

@@ -16,6 +16,7 @@
 from lfx.utils.flow_validation import (
     CustomComponentValidationError,
     prepare_public_flow_build,
+    validate_catalog_policy_for_flow,
     validate_flow_for_current_settings,
     validate_public_flow_no_code_execution,
 )
@@ -850,6 +851,13 @@ async def build_public_tmp(
         async with session_scope() as session:
             flow = await session.get(Flow, flow_id)
             if flow and flow.data:
+                # The default anonymous build path sanitizes component code directly
+                # and therefore does not call validate_flow_for_current_settings.
+                # Enforce the exact catalog snapshot after the public-access check
+                # and before any graph is queued or built. The explicit public-custom
+                # opt-in already runs the unified validator inside prepare_public_flow_build.
+                if not settings.allow_public_custom_components:
+                    validate_catalog_policy_for_flow(flow.data)
                 # Block unauthenticated builds of flows that run arbitrary code
                 # (Python interpreter/REPL, legacy Python Code Structured tool,
                 # Smart Transform lambda) or invoke another saved flow (Run Flow,

src/backend/base/langflow/api/v1/endpoints.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import time
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Collection
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Annotated, Any
 from uuid import UUID, uuid4
@@ -181,22 +181,34 @@ async def parse_input_request_from_body(http_request: Request) -> SimplifiedAPIR
 
 
 @router.get("/all")
-async def get_all(request: Request, current_user: CurrentActiveUser):
+async def get_all(request: Request, current_user: CurrentActiveUser, *, include_blocked: bool = False):
     """Retrieve all component types with compression for better performance.
 
     Returns a compressed response containing all available component types,
     with display_names translated to the locale indicated by Accept-Language.
     """
+    if include_blocked and not current_user.is_superuser:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only superusers can include blocked catalog components.",
+        )
+
     from langflow.interface.components import get_and_cache_all_types_dict
     from langflow.utils.i18n import build_component_display_names, translate_component_dict
 
     try:
+        catalog_policy_snapshot = get_catalog_policy_service().snapshot
         all_types_en = await get_and_cache_all_types_dict(settings_service=get_settings_service())
         visible_types_en = _filter_component_palette_by_provider_policy(
             all_types_en,
             user_id=current_user.id,
             attributes={"is_superuser": bool(current_user.is_superuser)},
         )
+        if not include_blocked:
+            visible_types_en = _filter_component_palette_by_catalog_policy(
+                visible_types_en,
+                blocked_component_keys=catalog_policy_snapshot.blocked_component_keys,
+            )
 
         locale = getattr(request.state, "locale", "en")
         all_types = translate_component_dict(visible_types_en, locale) if locale != "en" else visible_types_en
@@ -249,6 +261,28 @@ def _filter_component_palette_by_provider_policy(
     }
 
 
+def _filter_component_palette_by_catalog_policy(
+    all_types: dict[str, dict[str, dict]],
+    *,
+    blocked_component_keys: Collection[str],
+) -> dict[str, dict[str, dict]]:
+    """Return shallow category copies with exact blocked component keys removed.
+
+    Catalog policy keys match the inner component-registry keys exactly and
+    case-sensitively across every category. Category order, component order,
+    and empty categories are preserved. Component payloads remain shared with
+    the process-wide cache and are never mutated or deep-copied.
+    """
+    return {
+        category: {
+            component_key: component
+            for component_key, component in components.items()
+            if component_key not in blocked_component_keys
+        }
+        for category, components in all_types.items()
+    }
+
+
 def validate_input_and_tweaks(input_request: SimplifiedAPIRequest) -> None:
     # If the input_value is not None and the input_type is "chat"
     # then we need to check the tweaks if the ChatInput component is present

src/backend/base/langflow/api/v1/flows.py

@@ -4,6 +4,7 @@
 import io
 import threading
 import zipfile
+from collections.abc import Collection
 from typing import Annotated
 from uuid import UUID
 
@@ -49,7 +50,7 @@
 from langflow.api.v1.mappers.deployments.sync import retry_flow_operation_on_deployment_guard
 from langflow.api.v1.schemas import FlowListCreate
 from langflow.initial_setup.constants import STARTER_FOLDER_NAME
-from langflow.services.auth.utils import get_current_active_user
+from langflow.services.auth.utils import get_current_active_user, get_optional_user
 from langflow.services.authorization import (
     FlowAction,
     ensure_flow_permission,
@@ -78,7 +79,8 @@
 # and FlowVersionError from the flow_version modules.
 from langflow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
 from langflow.services.database.models.folder.model import Folder
-from langflow.services.deps import get_settings_service, get_storage_service
+from langflow.services.database.models.user.model import User
+from langflow.services.deps import get_catalog_policy_service, get_settings_service, get_storage_service
 from langflow.services.storage.service import StorageService
 from langflow.utils.compression import compress_response
 from langflow.utils.i18n import translate_flow_notes, translate_starter_flows
@@ -906,26 +908,60 @@ async def download_multiple_file(
 _starter_flows_lock = asyncio.Lock()
 
 
+def _filter_basic_examples_by_catalog_policy(
+    flows: list[FlowRead],
+    *,
+    blocked_template_keys: Collection[str],
+) -> list[FlowRead]:
+    """Return a request-local view without exact blocked template keys."""
+    return [flow for flow in flows if flow.name_key not in blocked_template_keys]
+
+
 @router.get("/basic_examples/", response_model=list[FlowRead], status_code=200)
 async def read_basic_examples(
     *,
     session: DbSession,
     request: Request,
+    user: Annotated[User | None, Depends(get_optional_user)],
+    include_blocked: bool = False,
 ):
     """Retrieve a list of basic example flows."""
+    if include_blocked and (user is None or not user.is_superuser):
+        raise HTTPException(
+            status_code=403,
+            detail="Only superusers can include blocked catalog templates.",
+        )
+
+    catalog_policy_snapshot = get_catalog_policy_service().snapshot
     locale = getattr(request.state, "locale", "en")
     translated_cache_key = f"starter_flows_{locale}"
 
     # Fast path: translated result already cached for this locale
     cached_translated = _starter_flows_translated_cache.get(translated_cache_key)
     if cached_translated is not CACHE_MISS:
-        return compress_response(cached_translated)
+        visible_flows = (
+            cached_translated
+            if include_blocked
+            else _filter_basic_examples_by_catalog_policy(
+                cached_translated,
+                blocked_template_keys=catalog_policy_snapshot.blocked_template_keys,
+            )
+        )
+        return compress_response(visible_flows)
 
     async with _starter_flows_lock:
         # Double-check inside lock to prevent thundering herd
         cached_translated = _starter_flows_translated_cache.get(translated_cache_key)
         if cached_translated is not CACHE_MISS:
-            return compress_response(cached_translated)
+            visible_flows = (
+                cached_translated
+                if include_blocked
+                else _filter_basic_examples_by_catalog_policy(
+                    cached_translated,
+                    blocked_template_keys=catalog_policy_snapshot.blocked_template_keys,
+                )
+            )
+            return compress_response(visible_flows)
 
         # Ensure raw DB data is cached
         cached_flow_reads = _starter_flows_cache.get("starter_flows")
@@ -967,7 +1003,15 @@ async def read_basic_examples(
 
         _starter_flows_translated_cache.set(translated_cache_key, result)
 
-    return compress_response(result)
+    visible_flows = (
+        result
+        if include_blocked
+        else _filter_basic_examples_by_catalog_policy(
+            result,
+            blocked_template_keys=catalog_policy_snapshot.blocked_template_keys,
+        )
+    )
+    return compress_response(visible_flows)
 
 
 @router.post("/expand/", status_code=200, dependencies=[Depends(get_current_active_user)], include_in_schema=False)

src/backend/base/langflow/api/v1/starter_projects.py

@@ -1,9 +1,10 @@
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, HTTPException, Request
 from pydantic import BaseModel
 
-from langflow.services.auth.utils import get_current_active_user
+from langflow.api.utils import CurrentActiveUser
+from langflow.services.deps import get_catalog_policy_service
 
 router = APIRouter(prefix="/starter-projects", tags=["Flows"])
 
@@ -37,16 +38,26 @@ class GraphDumpResponse(BaseModel):
     data: GraphData
     is_component: bool | None = None
     name: str | None = None
+    name_key: str
     description: str | None = None
     endpoint_name: str | None = None
 
 
-@router.get("/", dependencies=[Depends(get_current_active_user)], status_code=200)
-async def get_starter_projects(request: Request) -> list[GraphDumpResponse]:
+@router.get("/", status_code=200)
+async def get_starter_projects(
+    request: Request,
+    current_user: CurrentActiveUser,
+    *,
+    include_blocked: bool = False,
+) -> list[GraphDumpResponse]:
     """Get a list of starter projects."""
     from langflow.initial_setup.load import get_starter_projects_dump
     from langflow.utils.i18n import translate_flow_notes
 
+    if include_blocked and not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Only superusers can include blocked catalog templates.")
+
+    catalog_policy_snapshot = get_catalog_policy_service().snapshot
     locale = getattr(request.state, "locale", "en")
 
     try:
@@ -56,6 +67,12 @@ async def get_starter_projects(request: Request) -> list[GraphDumpResponse]:
         # Convert TypedDict GraphDump to Pydantic GraphDumpResponse
         results = []
         for item in raw_data:
+            name_key = item.get("name_key")
+            if not isinstance(name_key, str):
+                continue
+            if not include_blocked and catalog_policy_snapshot.is_template_blocked(name_key):
+                continue
+
             nodes = item.get("data", {}).get("nodes", [])
             translated_nodes = translate_flow_notes(nodes, locale)
 
@@ -71,6 +88,7 @@ async def get_starter_projects(request: Request) -> list[GraphDumpResponse]:
                 data=graph_data,
                 is_component=item.get("is_component"),
                 name=item.get("name"),
+                name_key=name_key,
                 description=item.get("description"),
                 endpoint_name=item.get("endpoint_name"),
             )

src/backend/base/langflow/initial_setup/load.py

@@ -1,6 +1,8 @@
 from collections.abc import Callable
 from typing import Any
 
+from langflow.utils.i18n_keys import safe_flow_key
+
 from .starter_projects import (
     basic_prompting_graph,
     blog_writer_graph,
@@ -56,5 +58,9 @@ def get_starter_projects_graphs():
 
 def get_starter_projects_dump():
     return [
-        build_graph().dump(name=name, description=description) for build_graph, name, description in STARTER_PROJECTS
+        {
+            **build_graph().dump(name=name, description=description),
+            "name_key": safe_flow_key(name),
+        }
+        for build_graph, name, description in STARTER_PROJECTS
     ]

src/backend/tests/unit/api/v1/test_flows.py

@@ -568,6 +568,71 @@ async def test_read_basic_examples(client: AsyncClient, logged_in_headers):
     assert response.status_code == status.HTTP_200_OK
     assert isinstance(result, list), "The result must be a list"
     assert len(result) > 0, "The result must have at least one flow"
+    assert all(item["name_key"] for item in result)
+
+
+async def test_read_basic_examples_catalog_policy_preserves_public_cache_and_unblocks(
+    client: AsyncClient,
+    monkeypatch,
+):
+    from langflow.api.v1 import flows
+    from lfx.services.catalog_policy import CatalogPolicySnapshot
+
+    class MutableCatalogPolicyService:
+        snapshot = CatalogPolicySnapshot(blocked_template_keys={"basic_prompting"})
+
+    service = MutableCatalogPolicyService()
+    monkeypatch.setattr(flows, "get_catalog_policy_service", lambda: service)
+    flows._starter_flows_cache.clear()
+    flows._starter_flows_translated_cache.clear()
+
+    blocked_response = await client.get("api/v1/flows/basic_examples/")
+    assert blocked_response.status_code == status.HTTP_200_OK, blocked_response.text
+    blocked_keys = {flow["name_key"] for flow in blocked_response.json()}
+    assert "basic_prompting" not in blocked_keys
+
+    service.snapshot = CatalogPolicySnapshot()
+    unblocked_response = await client.get("api/v1/flows/basic_examples/")
+    assert unblocked_response.status_code == status.HTTP_200_OK, unblocked_response.text
+    unblocked_keys = {flow["name_key"] for flow in unblocked_response.json()}
+    assert "basic_prompting" in unblocked_keys
+
+
+async def test_read_basic_examples_include_blocked_requires_superuser(
+    client: AsyncClient,
+    logged_in_headers,
+):
+    anonymous_response = await client.get("api/v1/flows/basic_examples/?include_blocked=true")
+    assert anonymous_response.status_code == status.HTTP_403_FORBIDDEN
+
+    denied_response = await client.get(
+        "api/v1/flows/basic_examples/?include_blocked=true",
+        headers=logged_in_headers,
+    )
+    assert denied_response.status_code == status.HTTP_403_FORBIDDEN
+
+
+async def test_read_basic_examples_superuser_can_include_blocked(
+    client: AsyncClient,
+    logged_in_headers_super_user,
+    monkeypatch,
+):
+    from langflow.api.v1 import flows
+    from lfx.services.catalog_policy import CatalogPolicySnapshot
+
+    service = type(
+        "CatalogPolicyService",
+        (),
+        {"snapshot": CatalogPolicySnapshot(blocked_template_keys={"basic_prompting"})},
+    )()
+    monkeypatch.setattr(flows, "get_catalog_policy_service", lambda: service)
+
+    override_response = await client.get(
+        "api/v1/flows/basic_examples/?include_blocked=true",
+        headers=logged_in_headers_super_user,
+    )
+    assert override_response.status_code == status.HTTP_200_OK, override_response.text
+    assert "basic_prompting" in {flow["name_key"] for flow in override_response.json()}
 
 
 async def test_read_flows_user_isolation(client: AsyncClient, logged_in_headers, active_user):