feat(catalog): expose governance config flag (#14319)

* feat(catalog): expose governance config flag

* feat(catalog): filter component palette by policy (#14320)

* feat(catalog): filter component palette by policy

* 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/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
@@ -80,6 +80,7 @@
 from langflow.services.database.models.user.model import User, UserRead
 from langflow.services.deps import (
     get_auth_service,
+    get_catalog_policy_service,
     get_job_service,
     get_memory_base_service,
     get_session_service,
@@ -180,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
@@ -248,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
@@ -1670,14 +1705,27 @@ async def get_config(
     """
     try:
         settings_service: SettingsService = get_settings_service()
+        try:
+            catalog_governance_enabled = get_catalog_policy_service().enabled
+        except Exception as exc:  # noqa: BLE001
+            # Catalog governance is explicitly fail-open. A broken custom
+            # policy implementation must not break the public config endpoint
+            # or expose its internal exception text.
+            await logger.aexception("Catalog policy status unavailable; reporting governance disabled", exception=exc)
+            catalog_governance_enabled = False
 
         if user is None:
             return PublicConfigResponse.from_settings(
                 settings_service.settings,
                 settings_service.auth_settings,
+                catalog_governance_enabled=catalog_governance_enabled,
             )
 
-        return ConfigResponse.from_settings(settings_service.settings, settings_service.auth_settings)
+        return ConfigResponse.from_settings(
+            settings_service.settings,
+            settings_service.auth_settings,
+            catalog_governance_enabled=catalog_governance_enabled,
+        )
 
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc

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/schemas/__init__.py

@@ -394,6 +394,9 @@ class BaseConfigResponse(BaseModel):
     # Mirrors ``LANGFLOW_AUTHZ_ENABLED``. EE/custom frontends gate the Access
     # Control settings entry on this flag; OSS UI ignores it until wired.
     authz_enabled: bool = False
+    # Signals that at least one component or template is governed without
+    # exposing the policy contents through the public config response.
+    catalog_governance_enabled: bool = False
 
 
 class PublicConfigResponse(BaseConfigResponse):
@@ -407,12 +410,19 @@ class PublicConfigResponse(BaseConfigResponse):
     allow_custom_components: bool
 
     @classmethod
-    def from_settings(cls, settings: Settings, auth_settings) -> "PublicConfigResponse":
+    def from_settings(
+        cls,
+        settings: Settings,
+        auth_settings,
+        *,
+        catalog_governance_enabled: bool = False,
+    ) -> "PublicConfigResponse":
         """Create a PublicConfigResponse instance using values from a Settings object.
 
         Parameters:
             settings (Settings): The Settings object containing configuration values.
             auth_settings: Auth settings (for ``authz_enabled``).
+            catalog_governance_enabled: Whether any catalog policy currently restricts resources.
 
         Returns:
             PublicConfigResponse: An instance populated with public-safe configuration values.
@@ -427,6 +437,7 @@ def from_settings(cls, settings: Settings, auth_settings) -> "PublicConfigRespon
             enable_extension_reload=settings.enable_extension_reload,
             allow_custom_components=settings.allow_custom_components,
             authz_enabled=bool(getattr(auth_settings, "AUTHZ_ENABLED", False)),
+            catalog_governance_enabled=catalog_governance_enabled,
         )
 
 
@@ -465,12 +476,19 @@ class ConfigResponse(BaseConfigResponse):
     agentic_experience: bool = True
 
     @classmethod
-    def from_settings(cls, settings: Settings, auth_settings) -> "ConfigResponse":
+    def from_settings(
+        cls,
+        settings: Settings,
+        auth_settings,
+        *,
+        catalog_governance_enabled: bool = False,
+    ) -> "ConfigResponse":
         """Create a ConfigResponse instance using values from a Settings object and AuthSettings.
 
         Parameters:
             settings (Settings): The Settings object containing configuration values.
             auth_settings: The AuthSettings object containing authentication configuration values.
+            catalog_governance_enabled: Whether any catalog policy currently restricts resources.
 
         Returns:
             ConfigResponse: An instance populated with configuration and feature flag values.
@@ -498,6 +516,7 @@ def from_settings(cls, settings: Settings, auth_settings) -> "ConfigResponse":
             hide_getting_started_progress=settings.hide_getting_started_progress,
             allow_custom_components=settings.allow_custom_components,
             authz_enabled=bool(getattr(auth_settings, "AUTHZ_ENABLED", False)),
+            catalog_governance_enabled=catalog_governance_enabled,
             embedded_mode=settings.embedded_mode,
             hide_logout_button=settings.hide_logout_button or settings.embedded_mode,
             hide_new_project_button=settings.hide_new_project_button or settings.embedded_mode,

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
     ]