feat(catalog): filter starter templates by policy (#14321)

* feat(catalog): filter starter templates by policy

* feat(catalog): enforce component policy at runtime (#14322)

Author: erichare

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/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):

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

@@ -2,6 +2,7 @@
 
 from fastapi import status
 from httpx import AsyncClient
+from lfx.services.catalog_policy import CatalogPolicySnapshot
 
 
 async def test_get_starter_projects(client: AsyncClient, logged_in_headers):
@@ -29,6 +30,14 @@ async def test_get_starter_projects(client: AsyncClient, logged_in_headers):
     "Vector Store RAG": "Load your data for chat context with Retrieval Augmented Generation.",
 }
 
+EXPECTED_STARTER_PROJECT_KEYS = {
+    "basic_prompting",
+    "blog_writer",
+    "document_q_a",
+    "memory_chatbot",
+    "vector_store_rag",
+}
+
 
 async def test_get_starter_projects_expose_canonical_metadata(client: AsyncClient, logged_in_headers):
     """Each starter project must expose its canonical name and description.
@@ -46,6 +55,7 @@ async def test_get_starter_projects_expose_canonical_metadata(client: AsyncClien
 
     names = [project["name"] for project in result]
     assert len(set(names)) == len(names), f"Starter project names must be unique, got: {names}"
+    assert {project["name_key"] for project in result} == EXPECTED_STARTER_PROJECT_KEYS
 
     actual = {project["name"]: project["description"] for project in result}
     assert actual == EXPECTED_STARTER_PROJECTS
@@ -58,6 +68,64 @@ async def test_get_starter_projects_expose_canonical_metadata(client: AsyncClien
     )
 
 
+async def test_starter_projects_catalog_policy_is_request_local_and_unblocks_without_restart(
+    client: AsyncClient,
+    logged_in_headers,
+    monkeypatch,
+):
+    from langflow.api.v1 import starter_projects
+
+    class MutableCatalogPolicyService:
+        snapshot = CatalogPolicySnapshot(blocked_template_keys={"basic_prompting"})
+
+    service = MutableCatalogPolicyService()
+    monkeypatch.setattr(starter_projects, "get_catalog_policy_service", lambda: service)
+
+    blocked_response = await client.get("api/v1/starter-projects/", headers=logged_in_headers)
+    assert blocked_response.status_code == status.HTTP_200_OK, blocked_response.text
+    assert {project["name_key"] for project in blocked_response.json()} == (
+        EXPECTED_STARTER_PROJECT_KEYS - {"basic_prompting"}
+    )
+
+    service.snapshot = CatalogPolicySnapshot()
+    unblocked_response = await client.get("api/v1/starter-projects/", headers=logged_in_headers)
+    assert unblocked_response.status_code == status.HTTP_200_OK, unblocked_response.text
+    assert {project["name_key"] for project in unblocked_response.json()} == EXPECTED_STARTER_PROJECT_KEYS
+
+
+async def test_starter_projects_include_blocked_requires_superuser(
+    client: AsyncClient,
+    logged_in_headers,
+):
+    denied_response = await client.get(
+        "api/v1/starter-projects/?include_blocked=true",
+        headers=logged_in_headers,
+    )
+    assert denied_response.status_code == status.HTTP_403_FORBIDDEN
+
+
+async def test_starter_projects_superuser_can_include_blocked(
+    client: AsyncClient,
+    logged_in_headers_super_user,
+    monkeypatch,
+):
+    from langflow.api.v1 import starter_projects
+
+    service = type(
+        "CatalogPolicyService",
+        (),
+        {"snapshot": CatalogPolicySnapshot(blocked_template_keys={"basic_prompting"})},
+    )()
+    monkeypatch.setattr(starter_projects, "get_catalog_policy_service", lambda: service)
+
+    override_response = await client.get(
+        "api/v1/starter-projects/?include_blocked=true",
+        headers=logged_in_headers_super_user,
+    )
+    assert override_response.status_code == status.HTTP_200_OK, override_response.text
+    assert {project["name_key"] for project in override_response.json()} == EXPECTED_STARTER_PROJECT_KEYS
+
+
 def test_starter_projects_keep_optional_crewai_exports_lazy():
     from langflow.initial_setup import starter_projects