feat(catalog): add policy service and admin API (#14318)

* feat(catalog): add policy service and admin API

* 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/router.py

@@ -14,6 +14,7 @@
     authz_roles_router,
     authz_shares_router,
     authz_teams_router,
+    catalog_policy_router,
     chat_router,
     endpoints_router,
     extensions_router,
@@ -101,6 +102,7 @@ def include_deployment_router(target_router: APIRouter) -> None:
 router_v1.include_router(authz_role_assignments_router)
 router_v1.include_router(authz_teams_router)
 router_v1.include_router(authz_me_router)
+router_v1.include_router(catalog_policy_router)
 
 
 # Extension reload is Mode A (local-dev / pip-installed) only.  The route is

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

@@ -6,6 +6,7 @@
 from langflow.api.v1.authz_roles import router as authz_roles_router
 from langflow.api.v1.authz_shares import router as authz_shares_router
 from langflow.api.v1.authz_teams import router as authz_teams_router
+from langflow.api.v1.catalog_policy import router as catalog_policy_router
 from langflow.api.v1.chat import router as chat_router
 from langflow.api.v1.endpoints import router as endpoints_router
 from langflow.api.v1.extensions import router as extensions_router
@@ -42,6 +43,7 @@
     "authz_roles_router",
     "authz_shares_router",
     "authz_teams_router",
+    "catalog_policy_router",
     "chat_router",
     "endpoints_router",
     "extensions_router",

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

@@ -0,0 +1,113 @@
+"""Superuser administration API for global catalog block policy."""
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+
+from langflow.api.v1.schemas.catalog_policy import CatalogPolicyBlockedSet
+from langflow.services.auth.utils import get_current_active_superuser
+from langflow.services.authorization.audit import audit_decision
+from langflow.services.database.models.user.model import User
+from langflow.services.deps import get_catalog_policy_service
+
+router = APIRouter(prefix="/catalog-policy", tags=["Catalog Policy"])
+
+
+def _response(blocked: frozenset[str]) -> CatalogPolicyBlockedSet:
+    return CatalogPolicyBlockedSet(blocked=sorted(blocked))
+
+
+async def _audit_update(
+    *,
+    user_id: UUID,
+    resource_kind: Literal["component", "template"],
+    added: frozenset[str],
+    removed: frozenset[str],
+) -> None:
+    """Emit one post-commit audit event per changed catalog key."""
+    for key in sorted(added):
+        await audit_decision(
+            user_id=user_id,
+            action="catalog:block",
+            obj=f"{resource_kind}:{key}",
+            result="allow",
+            details={
+                "resource_kind": resource_kind,
+                "resource_key": key,
+            },
+        )
+    for key in sorted(removed):
+        await audit_decision(
+            user_id=user_id,
+            action="catalog:unblock",
+            obj=f"{resource_kind}:{key}",
+            result="allow",
+            details={
+                "resource_kind": resource_kind,
+                "resource_key": key,
+            },
+        )
+
+
+@router.get("/components", response_model=CatalogPolicyBlockedSet)
+async def get_component_policy(
+    _admin: Annotated[User, Depends(get_current_active_superuser)],
+) -> CatalogPolicyBlockedSet:
+    """Return the complete global component block set."""
+    service = get_catalog_policy_service()
+    return _response(service.snapshot.blocked_component_keys)
+
+
+@router.put("/components", response_model=CatalogPolicyBlockedSet)
+async def replace_component_policy(
+    payload: CatalogPolicyBlockedSet,
+    admin: Annotated[User, Depends(get_current_active_superuser)],
+) -> CatalogPolicyBlockedSet:
+    """Replace the complete global component block set."""
+    service = get_catalog_policy_service()
+    update = await service.replace_blocked_component_keys(
+        payload.blocked,
+        actor_user_id=admin.id,
+    )
+    await _audit_update(
+        user_id=admin.id,
+        resource_kind="component",
+        added=update.added,
+        removed=update.removed,
+    )
+    return _response(update.snapshot.blocked_component_keys)
+
+
+@router.get("/templates", response_model=CatalogPolicyBlockedSet)
+async def get_template_policy(
+    _admin: Annotated[User, Depends(get_current_active_superuser)],
+) -> CatalogPolicyBlockedSet:
+    """Return the complete global template block set."""
+    service = get_catalog_policy_service()
+    return _response(service.snapshot.blocked_template_keys)
+
+
+@router.put("/templates", response_model=CatalogPolicyBlockedSet)
+async def replace_template_policy(
+    payload: CatalogPolicyBlockedSet,
+    admin: Annotated[User, Depends(get_current_active_superuser)],
+) -> CatalogPolicyBlockedSet:
+    """Replace the complete global template block set."""
+    service = get_catalog_policy_service()
+    update = await service.replace_blocked_template_keys(
+        payload.blocked,
+        actor_user_id=admin.id,
+    )
+    await _audit_update(
+        user_id=admin.id,
+        resource_kind="template",
+        added=update.added,
+        removed=update.removed,
+    )
+    return _response(update.snapshot.blocked_template_keys)
+
+
+__all__ = ["router"]

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)