fix: bind allowlisted extensions to discovered sources

CHANGELOG.md

@@ -13,6 +13,8 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Bind an allowlisted Python extension to its discovered entry-point source and
+  fail closed if its distribution identity changes before loading.
 - Preserve explicit application identities losslessly while using
   collision-resistant, path-safe runtime namespace components.
 - Give `BatteriesIncludedConfigLoader.cli_name` a documented identity role by

docs/extensions.md

@@ -76,3 +76,9 @@ Use `allowlist={"base_cli.commands:audit"}` to restrict names, or
 `ExtensionDiscovery(disabled=True)` to disable discovery entirely. Allowlist
 entries may be a bare entry-point name, a fully-qualified `group:name`, or a
 distribution name.
+
+Loading uses the exact entry-point source and distribution identity approved by
+discovery rather than re-selecting an entry point by a potentially colliding
+name/value tuple. If that identity changes before loading, the load fails
+closed. An allowlist is a selection boundary, not a sandbox for installed
+Python code.

lib/python/base_cli/extensions.py

@@ -183,6 +183,7 @@ def __init__(
         self._raw_cache: tuple[Any, ...] | None = None
         self._metadata_cache: tuple[ExtensionDescriptor, ...] | None = None
         self._descriptor_cache: dict[str, tuple[ExtensionDescriptor, ...]] = {}
+        self._entry_point_cache: dict[ExtensionDescriptor, Any] = {}
         self._loaded_cache: dict[tuple[str, str], Any] = {}
         self._lock = RLock()
 
@@ -268,6 +269,7 @@ def refresh(self) -> None:
             self._raw_cache = None
             self._metadata_cache = None
             self._descriptor_cache.clear()
+            self._entry_point_cache.clear()
             self._loaded_cache.clear()
 
     def _metadata_descriptors(self) -> tuple[ExtensionDescriptor, ...]:
@@ -288,6 +290,7 @@ def _metadata_descriptors(self) -> tuple[ExtensionDescriptor, ...]:
                 continue
             if self._allowed(descriptor):
                 descriptors.append(descriptor)
+                self._entry_point_cache[descriptor] = entry_point
         descriptors.sort(key=_descriptor_sort_key)
         self._metadata_cache = tuple(descriptors)
         return self._metadata_cache
@@ -319,14 +322,16 @@ def _allowed(self, descriptor: ExtensionDescriptor) -> bool:
         )
 
     def _load_descriptor(self, descriptor: ExtensionDescriptor) -> Any:
-        for entry_point in self._raw_entry_points():
-            if (
-                getattr(entry_point, "group", None) == descriptor.group
-                and getattr(entry_point, "name", None) == descriptor.name
-                and getattr(entry_point, "value", None) == descriptor.value
-            ):
-                return entry_point.load()
-        raise ImportError("entry point disappeared before it could be loaded")
+        entry_point = self._entry_point_cache.get(descriptor)
+        if entry_point is None:
+            raise ImportError("approved entry point disappeared before it could be loaded")
+        try:
+            current_descriptor = _descriptor_from_entry_point(entry_point)
+        except (AttributeError, TypeError, ValueError) as exc:
+            raise ImportError("approved entry point identity changed before it could be loaded") from exc
+        if current_descriptor != descriptor or not self._allowed(current_descriptor):
+            raise ImportError("approved entry point identity changed before it could be loaded")
+        return entry_point.load()
 
 
 def _validate_group(group: str) -> None:

tests/test_extensions.py

@@ -76,6 +76,41 @@ def test_allowlist_and_disable_switch_are_enforced(self) -> None:
         with self.assertRaises(base_cli.ExtensionsDisabledError):
             disabled.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "allowed")
 
+    def test_allowlist_loads_the_exact_approved_source_across_collisions_and_refresh(self) -> None:
+        for trusted_first in (False, True):
+            with self.subTest(trusted_first=trusted_first):
+                denied = _entry_point("shared", "same.module:plugin", distribution="denied")
+                trusted = _entry_point("shared", "same.module:plugin", distribution="trusted")
+                denied.load = mock.Mock(return_value="denied")
+                trusted.load = mock.Mock(return_value="trusted")
+                entries = (trusted, denied) if trusted_first else (denied, trusted)
+                discovery = base_cli.ExtensionDiscovery(entry_points=entries, allowlist={"trusted"})
+
+                descriptors = discovery.list_commands()
+                self.assertEqual(len(descriptors), 1)
+                self.assertEqual(descriptors[0].distribution, "trusted")
+                self.assertEqual(discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "shared"), "trusted")
+                result = discovery.load_all(base_cli.COMMAND_ENTRY_POINT_GROUP)
+                self.assertEqual([item.value for item in result], ["trusted"])
+                denied.load.assert_not_called()
+                trusted.load.assert_called_once_with()
+
+                discovery.refresh()
+                self.assertEqual(discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "shared"), "trusted")
+                denied.load.assert_not_called()
+                trusted.load.assert_has_calls([mock.call(), mock.call()])
+
+    def test_allowlisted_entry_point_identity_change_fails_closed(self) -> None:
+        trusted = _entry_point("shared", "same.module:plugin", distribution="trusted")
+        trusted.load = mock.Mock(return_value="trusted")
+        discovery = base_cli.ExtensionDiscovery(entry_points=(trusted,), allowlist={"trusted"})
+        self.assertEqual(len(discovery.list_commands()), 1)
+
+        trusted.dist.name = "denied"
+        with self.assertRaisesRegex(base_cli.ExtensionLoadError, "identity changed"):
+            discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "shared")
+        trusted.load.assert_not_called()
+
     def test_malformed_metadata_is_skipped_without_hiding_healthy_extensions(self) -> None:
         malformed = _entry_point(
             "broken",