security(plugins): govern ALL entry-point groups by the unified allow/block (total trust boundary)

fluid_build/cli/apply.py

@@ -142,11 +142,19 @@ def _run_apply_hooks(
         except TypeError:
             eps = _md.entry_points().get("fluid_build.apply_hooks", [])
     except Exception as e:
-        logger.warning("apply hook discovery failed: %s", redact_secret_text(str(e)))
+        logger.warning("apply hook discovery failed: %s", type(e).__name__)
         return 0
 
+    from fluid_build.plugin_manager import is_allowed
+
     errors: List[str] = []
     for ep in eps:
+        # Operator allow/block gate — an apply_hook runs plugin code, so the same
+        # FLUID_PLUGINS_ALLOWLIST/BLOCKLIST policy that governs every other plugin
+        # surface governs it too. Enforced BEFORE ep.load().
+        if not is_allowed(ep.name):
+            logger.debug("apply hook %s skipped by allow/block policy", ep.name)
+            continue
         # Defense-in-depth: each hook gets its own deep copy. A hook can
         # observe the contract freely but cannot poison the data structure
         # the rest of apply or other hooks rely on.

fluid_build/cli/bootstrap.py

@@ -26,7 +26,6 @@
 from typing import Any, Dict, List, Optional, Tuple
 
 from fluid_build.cli.console import error as console_error
-from fluid_build.observability.secret_redactor import redact_secret_text
 
 # Import shared utilities
 from ._common import build_provider, load_contract_with_overlay
@@ -607,11 +606,10 @@ def register_core_commands(sp: argparse._SubParsersAction) -> None:
     # call ``add_parser`` on it. Failures are logged at WARNING level so a
     # broken plugin can never break ``fluid`` itself.
     # ────────────────────────────────────────────────────────────────────
-    # ``redact_secret_text`` is imported at module top, so both the inner
-    # plugin-load path and the outer discovery-failure path are guaranteed
-    # to scrub plugin-supplied exception text before logging — there is no
-    # code path where a credential-shaped substring from ``str(exception)``
-    # reaches the log handler without going through the redactor.
+    # Governance: each command plugin is gated by the operator allow/block
+    # policy (``is_allowed``) BEFORE it is loaded, and load/discovery failures
+    # are logged by exception TYPE only — plugin-supplied exception text never
+    # reaches the log handler, so a credential-shaped substring cannot leak.
     try:
         import importlib.metadata as _md
 
@@ -620,14 +618,16 @@ def register_core_commands(sp: argparse._SubParsersAction) -> None:
         except TypeError:
             # Python < 3.10 returned a dict of groups, not a kwarg-filtered iter.
             _eps = _md.entry_points().get("fluid_build.commands", [])
+        from fluid_build.plugin_manager import is_allowed
+
         for _ep in _eps:
+            if not is_allowed(_ep.name):
+                LOG.debug("CLI command plugin %s skipped by allow/block policy", _ep.name)
+                continue
             try:
                 _ep.load()(sp)
             except Exception as _e:
-                LOG.warning(
-                    "Failed to load CLI plugin %s: %s",
-                    _ep.name,
-                    redact_secret_text(str(_e)),
-                )
+                # Type-only — never interpolate plugin-supplied exception text.
+                LOG.warning("Failed to load CLI plugin %s: %s", _ep.name, type(_e).__name__)
     except Exception as _e:
-        LOG.warning("CLI plugin discovery failed: %s", redact_secret_text(str(_e)))
+        LOG.warning("CLI plugin discovery failed: %s", type(_e).__name__)

fluid_build/copilot/catalog/source_registry.py

@@ -129,7 +129,7 @@ def _discover_entrypoints(logger: Optional[logging.Logger]) -> None:
         else:  # Python < 3.10
             eps = all_eps.get(EP_GROUP, [])
     except Exception as exc:  # noqa: BLE001 — discovery itself failed; fail open
-        log.warning("source adapter discovery failed: %s", exc)
+        log.warning("source adapter discovery failed: %s", type(exc).__name__)
         return
 
     for ep in eps:
@@ -231,5 +231,13 @@ def resolve_catalog_adapter_class(name: str) -> Any:
         module_path, class_name = target.split(":", 1)
         module = __import__(module_path, fromlist=[class_name])
         return getattr(module, class_name)
-    # plugin EntryPoint — load lazily
+    # Third-party plugin EntryPoint — gate by the operator allow/block policy
+    # BEFORE the lazy load, so a blocked source adapter's code never executes.
+    from fluid_build.plugin_manager import is_allowed
+
+    if not is_allowed(key):
+        raise RuntimeError(
+            f"Source adapter {name!r} is blocked by the operator allow/block policy "
+            f"(FLUID_PLUGINS_ALLOWLIST / FLUID_PLUGINS_BLOCKLIST)."
+        )
     return target.load()

fluid_build/copilot/modeling_techniques.py

@@ -147,7 +147,7 @@ def _discover_entrypoints(logger: Optional[logging.Logger]) -> None:
             else all_eps.get(EP_GROUP, [])
         )
     except Exception as exc:  # noqa: BLE001 — fail open
-        log.warning("modeling-technique discovery failed: %s", exc)
+        log.warning("modeling-technique discovery failed: %s", type(exc).__name__)
         return
 
     for ep in eps:
@@ -159,10 +159,16 @@ def _discover_entrypoints(logger: Optional[logging.Logger]) -> None:
                 "modeling-technique plugin %r shadows a built-in; keeping the built-in", name
             )
             continue
+        from fluid_build.plugin_manager import is_allowed
+
+        if not is_allowed(name):
+            log.debug("modeling-technique plugin %r skipped by allow/block policy", name)
+            continue
         try:
             obj = ep.load()
         except Exception as exc:  # noqa: BLE001 — one broken plugin must not break the CLI
-            log.warning("modeling-technique plugin %r failed to load: %s", name, exc)
+            # Type-only — never interpolate plugin-supplied exception text.
+            log.warning("modeling-technique plugin %r failed to load: %s", name, type(exc).__name__)
             continue
         if not isinstance(obj, ModelingTechnique):
             log.warning(

fluid_build/extension_schemas.py

@@ -72,8 +72,13 @@ def iter_extension_schemas(
         log.warning("Extension schema discovery failed: %s", redact_secret_text(str(e)))
         return {}
 
+    from fluid_build.plugin_manager import is_allowed
+
     schemas: Dict[str, Dict[str, Any]] = {}
     for ep in eps:
+        if not is_allowed(ep.name):
+            log.debug("extension schema provider %r skipped by allow/block policy", ep.name)
+            continue
         try:
             provider = ep.load()
             try:
@@ -127,7 +132,12 @@ def run_extension_validators(
         return []
 
     errors: List[str] = []
+    from fluid_build.plugin_manager import is_allowed
+
     for ep in eps:
+        if not is_allowed(ep.name):
+            log.debug("extension validator %r skipped by allow/block policy", ep.name)
+            continue
         plugin_errors: List[str] = []
         try:
             validator = ep.load()

fluid_build/plugin_manager.py

@@ -55,6 +55,31 @@
     "iac_provider": "fluid_build.iac_providers",
 }
 
+# CLI-internal entry-point groups that also load + run plugin code. They are not
+# fluid_sdk *roles*, but they ARE operator-governable plugin surfaces, so they are
+# subject to the SAME allow/block policy and surfaced by ``fluid plugins``. Every
+# entry-point group whose plugins the CLI executes belongs in exactly one of
+# ROLE_GROUPS / EXTRA_GROUPS so the allow/block + type-only guarantees are total.
+EXTRA_GROUPS: Dict[str, str] = {
+    "command": "fluid_build.commands",
+    "apply_hook": "fluid_build.apply_hooks",
+    "extension_schema": "fluid_build.extension_schemas",
+    "extension_validator": "fluid_build.extension_validators",
+    "modeling_technique": "fluid_build.modeling_techniques",
+    "source_adapter": "fluid_build.source_adapters",
+}
+
+
+def governed_groups() -> Dict[str, str]:
+    """Return every entry-point group the operator allow/block policy governs.
+
+    ``ROLE_GROUPS`` (the fluid_sdk roles) plus ``EXTRA_GROUPS`` (CLI-internal
+    plugin surfaces). Anything here is gated by :func:`is_allowed` at its walk
+    site and listed by ``fluid plugins``.
+    """
+    return {**ROLE_GROUPS, **EXTRA_GROUPS}
+
+
 _ALLOWLIST_ENV = "FLUID_PLUGINS_ALLOWLIST"
 _BLOCKLIST_ENV = "FLUID_PLUGINS_BLOCKLIST"
 
@@ -119,20 +144,22 @@ def list_plugins(role: Optional[str] = None) -> Dict[str, List[str]]:
 
 
 def installed_plugins(role: Optional[str] = None) -> Dict[str, List[Dict[str, Any]]]:
-    """Return ``{role: [{name, group, allowed}]}`` for ALL installed plugins.
+    """Return ``{group_key: [{name, group, allowed}]}`` for ALL installed plugins.
 
     Unlike :func:`list_plugins` (which filters to the allowed set), this surfaces
-    EVERY installed plugin per role with its allow/block status, for the operator
-    inspection command (``fluid plugins``). Reads entry-point *names* only — it
-    never imports plugin code.
+    EVERY installed plugin in EVERY governed group — the fluid_sdk roles AND the
+    CLI-internal groups (commands / apply_hooks / extension_*) — with its
+    allow/block status, for the operator inspection command (``fluid plugins``).
+    Reads entry-point *names* only — it never imports plugin code.
     """
-    roles = [role] if role else list(ROLE_GROUPS)
+    groups = governed_groups()
+    keys = [role] if role else list(groups)
     out: Dict[str, List[Dict[str, Any]]] = {}
-    for r in roles:
-        group = ROLE_GROUPS.get(r)
+    for k in keys:
+        group = groups.get(k)
         if not group:
             continue
-        out[r] = [
+        out[k] = [
             {"name": ep.name, "group": group, "allowed": is_allowed(ep.name)}
             for ep in sorted(_entry_points(group), key=lambda e: e.name)
         ]
@@ -270,6 +297,8 @@ def dispatch_catalog_adapters(
 
 __all__ = [
     "ROLE_GROUPS",
+    "EXTRA_GROUPS",
+    "governed_groups",
     "is_allowed",
     "iter_plugins",
     "list_plugins",

tests/test_governed_groups.py

@@ -0,0 +1,147 @@
+# Copyright 2024-2026 Agentics Transformation Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Every entry-point group the CLI executes is governed by ONE allow/block policy.
+
+Previously six CLI-internal groups (commands / apply_hooks / extension_schemas /
+extension_validators / modeling_techniques / source_adapters) bypassed the gate
+and were invisible to `fluid plugins`.
+"""
+
+from __future__ import annotations
+
+import importlib.metadata as md
+
+from fluid_build import extension_schemas as ES
+from fluid_build import plugin_manager as PM
+
+
+class _FakeEP:
+    def __init__(self, name, obj):
+        self.name = name
+        self._obj = obj
+
+    def load(self):
+        return self._obj
+
+
+# ── governed-group surface ────────────────────────────────────────────
+
+
+def test_governed_groups_covers_roles_and_cli_internal():
+    g = PM.governed_groups()
+    # the five SDK roles + the four CLI-internal groups
+    assert set(PM.ROLE_GROUPS) <= set(g)
+    assert set(PM.EXTRA_GROUPS) <= set(g)
+    assert g["apply_hook"] == "fluid_build.apply_hooks"
+    assert g["extension_validator"] == "fluid_build.extension_validators"
+
+
+def test_installed_plugins_surfaces_cli_internal_groups(monkeypatch):
+    monkeypatch.setattr(PM, "_entry_points", lambda group: [])
+    keys = set(PM.installed_plugins())
+    # `fluid plugins` now sees commands / apply_hooks / extension_* too.
+    assert {"command", "apply_hook", "extension_schema", "extension_validator"} <= keys
+
+
+# ── extension_schemas / extension_validators honour allow/block ───────
+
+
+def test_extension_schema_provider_blocked_is_not_loaded(monkeypatch):
+    loaded = {"called": False}
+
+    def _provider(fluid_version=None):
+        loaded["called"] = True
+        return {"type": "object"}
+
+    monkeypatch.setattr(md, "entry_points", lambda **kw: [_FakeEP("myext", _provider)])
+    monkeypatch.setenv("FLUID_PLUGINS_BLOCKLIST", "myext")
+    monkeypatch.delenv("FLUID_PLUGINS_ALLOWLIST", raising=False)
+    schemas = ES.iter_extension_schemas()
+    assert "myext" not in schemas
+    assert loaded["called"] is False  # blocked BEFORE load — code never ran
+
+
+def test_extension_schema_provider_allowed_loads(monkeypatch):
+    monkeypatch.setattr(
+        md,
+        "entry_points",
+        lambda **kw: [_FakeEP("myext", lambda fluid_version=None: {"type": "object"})],
+    )
+    monkeypatch.delenv("FLUID_PLUGINS_BLOCKLIST", raising=False)
+    monkeypatch.delenv("FLUID_PLUGINS_ALLOWLIST", raising=False)
+    assert "myext" in ES.iter_extension_schemas()
+
+
+def test_extension_validator_blocked_does_not_run(monkeypatch):
+    ran = {"called": False}
+
+    def _validator(extensions, errors):
+        ran["called"] = True
+        errors.append("nope")
+
+    monkeypatch.setattr(md, "entry_points", lambda **kw: [_FakeEP("myext", _validator)])
+    monkeypatch.setenv("FLUID_PLUGINS_BLOCKLIST", "myext")
+    monkeypatch.delenv("FLUID_PLUGINS_ALLOWLIST", raising=False)
+    errors = ES.run_extension_validators({"extensions": {"myext": {}}})
+    assert ran["called"] is False  # blocked before load
+    assert errors == []
+
+
+# ── source_adapters: lazy load gated by allow/block ───────────────────
+
+
+def test_source_adapter_blocked_before_lazy_load(monkeypatch):
+    import pytest
+
+    from fluid_build.copilot.catalog import source_registry as SR
+
+    class _Target:
+        def load(self):
+            raise AssertionError("must not load a blocked source adapter")
+
+    monkeypatch.setattr(SR, "_ensure_discovered", lambda: None)
+    monkeypatch.setitem(
+        SR._REGISTRY,
+        "mysrc",
+        SR.SourceAdapterSpec(name="mysrc", kind="catalog", target=_Target(), origin="plugin"),
+    )
+    monkeypatch.setenv("FLUID_PLUGINS_BLOCKLIST", "mysrc")
+    monkeypatch.delenv("FLUID_PLUGINS_ALLOWLIST", raising=False)
+    with pytest.raises(RuntimeError, match="blocked by the operator allow/block"):
+        SR.resolve_catalog_adapter_class("mysrc")
+
+
+# ── modeling_techniques: entry-point load gated by allow/block ────────
+
+
+def test_modeling_technique_blocked_before_load(monkeypatch):
+    import importlib.metadata as _md
+
+    from fluid_build.copilot import modeling_techniques as MT
+
+    loaded = {"called": False}
+
+    def _loader():
+        loaded["called"] = True
+        return MT.ModelingTechnique(name="mytech", description="x", origin="plugin")
+
+    monkeypatch.setattr(
+        _md, "entry_points", lambda **kw: {MT.EP_GROUP: [_FakeEP("mytech", _loader)]}
+    )
+    monkeypatch.setenv("FLUID_PLUGINS_BLOCKLIST", "mytech")
+    monkeypatch.delenv("FLUID_PLUGINS_ALLOWLIST", raising=False)
+    MT._discover_entrypoints(None)
+    assert "mytech" not in MT._REGISTRY  # blocked → not registered
+    assert loaded["called"] is False  # and never loaded

tests/test_plugins_cmd.py

@@ -42,10 +42,11 @@ def test_installed_plugins_reports_allow_block_status(monkeypatch):
     assert by_name["good"]["group"] == "fluid_build.validators"
 
 
-def test_installed_plugins_covers_all_roles(monkeypatch):
+def test_installed_plugins_covers_all_governed_groups(monkeypatch):
     monkeypatch.setattr(PM, "_entry_points", lambda group: [])
     data = PM.installed_plugins()
-    assert set(data) == set(PM.ROLE_GROUPS)
+    # Now covers the SDK roles AND the CLI-internal governed groups.
+    assert set(data) == set(PM.governed_groups())
 
 
 # ── the `fluid plugins` command is wired into the CLI ─────────────────