Make whitelist_modules only apply to remote dispatch

salt/loader/__init__.py

@@ -358,13 +358,41 @@ def minion_mods(
     # TODO Publish documentation for module whitelisting
     if not whitelist:
         whitelist = opts.get("whitelist_modules", None)
+    # Both loaders must share the same ``__context__`` dict.  If we
+    # leave it as ``None`` LazyLoader.__init__ replaces it with a fresh
+    # ``{}`` in each loader's ``self.pack``, so writes made via one
+    # loader's NamedLoaderContext never reach reads made via the other's.
+    # Materialising the dict here keeps both packs pointing at the same
+    # object.
+    if context is None:
+        context = {}
     pack = {
         "__context__": context,
         "__utils__": utils,
         "__proxy__": proxy,
         "__opts__": opts,
         "__file_client__": file_client,
     }
+    # Two-loader model: outer loader is whitelist-filtered for wire
+    # dispatch; inner ``salt_dunder`` is unfiltered and packed as
+    # ``__salt__`` inside every loaded module, so a whitelisted module
+    # can still compose with non-whitelisted modules via ``__salt__[...]``.
+    # When no whitelist is set both loaders load the same set of modules;
+    # LazyLoader reuses an existing per-module ``LoaderContext`` when it
+    # encounters one, so both loaders share the same NamedLoaderContext
+    # bindings and per-module ``__context__`` state stays consistent.
+    salt_dunder = LazyLoader(
+        _module_dirs(opts, "modules", "module"),
+        opts,
+        tag="module",
+        pack=pack,
+        loaded_base_name=loaded_base_name,
+        static_modules=static_modules,
+        extra_module_dirs=utils.module_dirs if utils else None,
+        pack_self="__salt__",
+    )
+    pack = dict(pack)
+    pack["__salt__"] = salt_dunder
     if pillar is not None:
         pack["__pillar__"] = pillar
     ret = LazyLoader(
@@ -376,12 +404,11 @@ def minion_mods(
         loaded_base_name=loaded_base_name,
         static_modules=static_modules,
         extra_module_dirs=utils.module_dirs if utils else None,
-        pack_self="__salt__",
     )
 
     # Allow the usage of salt dunder in utils modules.
     if utils and isinstance(utils, LazyLoader):
-        utils.pack["__salt__"] = ret
+        utils.pack["__salt__"] = salt_dunder
 
     # Load any provider overrides from the configuration file providers option
     #  Note: Providers can be pkg, service, user or group - not to be confused

salt/modules/saltcheck.py

@@ -298,15 +298,22 @@
 
 log = logging.getLogger(__name__)
 
-try:
-    __context__
-except NameError:
-    __context__ = {}
-__context__["global_scheck"] = None
-
 __virtualname__ = "saltcheck"
 
 
+def __init__(opts):
+    # Initialise ``global_scheck`` in the loader's ``__context__`` on
+    # every load, but only if no previous load has already populated it.
+    # Doing this at module top-level would be unsafe: module-level code
+    # runs *before* the loader's pack loop binds ``__context__`` to the
+    # loader's ``NamedLoaderContext``, so a fresh dict created there is
+    # orphaned when the pack loop rewires ``__context__``.  It would
+    # also unconditionally reset the entry on every ``exec_module``,
+    # clobbering the ``SaltCheck`` instance a running call has already
+    # stored.
+    __context__.setdefault("global_scheck", None)
+
+
 def __virtual__():
     """
     Set the virtual pkg module if not running as a proxy

salt/utils/optsdict.py

@@ -462,6 +462,10 @@ def __init__(
         self._base = base_dict if base_dict is not None else {}
         self._name = name or f"OptsDict@{id(self)}"
         self._lock = threading.RLock()
+        # Cache of {key: (proxy, id(underlying_value))} to avoid re-allocating
+        # a DictProxy/ListProxy on every read of the same mutable value.
+        # Invalidated on __setitem__/__delitem__/COW (id changes).
+        self._proxy_cache: dict[str, tuple[Any, int]] = {}
 
         # Mutation tracking
         if parent and parent._tracker:
@@ -541,7 +545,8 @@ def __getitem__(self, key: str) -> Any:
 
         When accessing mutable values from parent/base, we return a proxy object
         that triggers copy-on-write on first mutation. This provides isolation
-        without copying until actually needed.
+        without copying until actually needed. Proxies are cached per key so
+        repeated reads of the same underlying value don't reallocate.
         """
         with self._ensure_lock():
             # Check local first - if already copied, return direct reference
@@ -561,9 +566,9 @@ def __getitem__(self, key: str) -> Any:
                         raise KeyError(key)
                     # Wrap mutable values in proxies to catch mutations
                     if isinstance(value, dict) and not isinstance(value, OptsDict):
-                        return DictProxy(value, self, key)
+                        return self._proxy_for(key, value, DictProxy)
                     elif isinstance(value, list):
-                        return ListProxy(value, self, key)
+                        return self._proxy_for(key, value, ListProxy)
                     # Immutable values can be returned directly
                     return value
 
@@ -573,13 +578,27 @@ def __getitem__(self, key: str) -> Any:
                 # Even root instances need proxies to track when values are mutated
                 # This allows us to know when a key has been accessed/modified
                 if isinstance(value, dict) and not isinstance(value, OptsDict):
-                    return DictProxy(value, self, key)
+                    return self._proxy_for(key, value, DictProxy)
                 elif isinstance(value, list):
-                    return ListProxy(value, self, key)
+                    return self._proxy_for(key, value, ListProxy)
                 return value
 
             raise KeyError(key)
 
+    def _proxy_for(self, key: str, value: Any, cls: type) -> Any:
+        """
+        Return a cached proxy for ``value`` at ``key``, allocating a new one
+        only when the underlying object identity has changed.
+        """
+        entry = self._proxy_cache.get(key)
+        if entry is not None:
+            proxy, cached_id = entry
+            if cached_id == id(value):
+                return proxy
+        proxy = cls(value, self, key)
+        self._proxy_cache[key] = (proxy, id(value))
+        return proxy
+
     def __setitem__(self, key: str, value: Any):
         """
         Set item with copy-on-write semantics.
@@ -606,6 +625,10 @@ def __setitem__(self, key: str, value: Any):
                 # Subsequent mutation of already-local key
                 self._tracker.record_mutation(key, original_value, value)
 
+            # Invalidate any cached proxy for this key: the underlying value
+            # is changing, so a re-read must not hand back a proxy pointing
+            # at the stale target.
+            self._proxy_cache.pop(key, None)
             # Store the value locally
             self._local[key] = value
 
@@ -635,6 +658,9 @@ def __delitem__(self, key: str):
             if key not in self:
                 raise KeyError(key)
 
+            # Invalidate any cached proxy for this key.
+            self._proxy_cache.pop(key, None)
+
             if key in self._local:
                 # Key is in local - check if it's already deleted
                 if self._local[key] is _DELETED:

tests/pytests/integration/loader/test_module_whitelist_dunder.py

@@ -0,0 +1,139 @@
+"""
+Integration tests for the split-loader behavior in ``salt.loader.minion_mods``.
+
+``minion_mods()`` returns a whitelist-filtered LazyLoader for remote
+dispatch, but packs an *unfiltered* loader as ``__salt__`` inside every
+loaded module.
+
+Effect on a whitelisted minion:
+  - Remote publishers can only invoke functions from whitelisted modules.
+  - A whitelisted module can still compose with non-whitelisted modules
+    via ``__salt__[...]``.
+"""
+
+import pytest
+
+from tests.conftest import FIPS_TESTRUN
+
+SECTEST_MODULE = """
+def run(cmd):
+    return __salt__["cmd.run"](cmd)
+"""
+
+
+@pytest.fixture
+def whitelisted_minion(salt_master):
+    """
+    A minion configured with ``whitelist_modules: [test, sectest, saltutil]``.
+    ``cmd`` is *deliberately absent* from the whitelist.
+    """
+    minion = salt_master.salt_minion_daemon(
+        "test-whitelist-dunder-minion",
+        overrides={
+            "whitelist_modules": [
+                "test",
+                "sectest",
+                "saltutil",
+                # Needed for the SLS-render tests below (state.template_str
+                # touches config/grains/pillar/slsutil during compilation).
+                "state",
+                "config",
+                "grains",
+                "pillar",
+                "slsutil",
+            ],
+            "fips_mode": FIPS_TESTRUN,
+            "encryption_algorithm": "OAEP-SHA224" if FIPS_TESTRUN else "OAEP-SHA1",
+            "signing_algorithm": (
+                "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1"
+            ),
+        },
+    )
+    minion.after_terminate(
+        pytest.helpers.remove_stale_minion_key, salt_master, minion.id
+    )
+    with salt_master.state_tree.base.temp_file("_modules/sectest.py", SECTEST_MODULE):
+        with minion.started():
+            salt_cli = salt_master.salt_cli()
+            salt_cli.run("saltutil.sync_modules", minion_tgt=minion.id)
+            yield minion
+
+
+def test_whitelisted_function_returns(salt_cli, whitelisted_minion):
+    """
+    ``test.ping`` is on the whitelist and must return normally.
+    """
+    ret = salt_cli.run("test.ping", minion_tgt=whitelisted_minion.id)
+    assert ret.data is True
+
+
+def test_nonwhitelisted_function_is_blocked(salt_cli, whitelisted_minion):
+    """
+    ``cmd.run`` is *not* on the whitelist. Remote publish must not
+    execute it: the minion's outer (filtered) loader has no ``cmd``
+    entry, so the function is unavailable and the CLI reports either
+    "'cmd.run' is not available." or "Minion did not return" -- both
+    prove the whitelist rejected the call.
+    """
+    ret = salt_cli.run(
+        "cmd.run", "echo blocked", minion_tgt=whitelisted_minion.id, _timeout=15
+    )
+    data = str(ret.data or "")
+    assert "not available" in data or "did not return" in data
+
+
+def test_whitelisted_module_reaches_nonwhitelisted_via_dunder(
+    salt_cli, whitelisted_minion
+):
+    """
+    ``sectest`` is whitelisted; its ``run()`` internally calls
+    ``__salt__['cmd.run']``. Because the packed ``__salt__`` is the
+    *unfiltered* loader, the call succeeds even though direct remote
+    dispatch of ``cmd.run`` is blocked (previous test).
+    """
+    ret = salt_cli.run(
+        "sectest.run", "echo hello-from-dunder", minion_tgt=whitelisted_minion.id
+    )
+    assert ret.data == "hello-from-dunder"
+
+
+def test_sls_render_can_call_whitelisted_module(salt_cli, whitelisted_minion):
+    """
+    SLS files render on the minion with the whitelist-filtered loader
+    exposed as ``salt`` / ``__salt__``. A whitelisted module call inside
+    the template must render normally and the resulting state must run.
+    """
+    template = (
+        "{% set r = salt['test.echo']('hi-from-sls') %}\n"
+        "probe:\n"
+        "  test.nop:\n"
+        "    - name: {{ r }}\n"
+    )
+    ret = salt_cli.run("state.template_str", template, minion_tgt=whitelisted_minion.id)
+    # state.template_str returns a dict keyed by state chunk id.
+    assert isinstance(ret.data, dict)
+    key = next(iter(ret.data))
+    assert ret.data[key]["result"] is True
+    assert ret.data[key]["name"] == "hi-from-sls"
+
+
+def test_sls_render_cannot_call_nonwhitelisted_module(salt_cli, whitelisted_minion):
+    """
+    ``cmd`` is not on ``whitelist_modules``. A template that tries
+    ``salt['cmd.run'](...)`` must fail *at render time* -- the render
+    pipeline receives the same filtered loader that the wire dispatch
+    uses, not the unfiltered ``salt_dunder`` that execution modules see.
+
+    Jinja surfaces the missing key as ``UndefinedError: '...AliasedLoader
+    object' has no attribute 'cmd.run'``.
+    """
+    template = (
+        "{% set r = salt['cmd.run']('id') %}\n"
+        "probe:\n"
+        "  test.nop:\n"
+        "    - name: {{ r }}\n"
+    )
+    ret = salt_cli.run("state.template_str", template, minion_tgt=whitelisted_minion.id)
+    text = str(ret.data or ret.stdout)
+    assert "cmd.run" in text
+    assert "UndefinedError" in text or "no attribute" in text