diff --git a/CHANGELOG.md b/CHANGELOG.md index ec20c2e..57ea474 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/extensions.md b/docs/extensions.md index 46f120a..a58d8fb 100644 --- a/docs/extensions.md +++ b/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. diff --git a/lib/python/base_cli/extensions.py b/lib/python/base_cli/extensions.py index 880713e..f3599c7 100644 --- a/lib/python/base_cli/extensions.py +++ b/lib/python/base_cli/extensions.py @@ -8,6 +8,7 @@ from __future__ import annotations import importlib.metadata as metadata +import zipfile from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass from pathlib import Path @@ -183,6 +184,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() @@ -223,25 +225,24 @@ def load(self, group: str, name: str) -> Any: if self.disabled: raise ExtensionsDisabledError("Python extension discovery is disabled") _validate_group(group) - matches = tuple(descriptor for descriptor in self.list(group) if descriptor.name == name) - if not matches: - raise ExtensionDiscoveryError(f"No extension named '{name}' exists in group '{group}'.") - if len(matches) > 1: - raise ExtensionCollisionError(group, name, matches) - key = (group, name) with self._lock: + matches = tuple(descriptor for descriptor in self.list(group) if descriptor.name == name) + if not matches: + raise ExtensionDiscoveryError(f"No extension named '{name}' exists in group '{group}'.") + if len(matches) > 1: + raise ExtensionCollisionError(group, name, matches) + key = (group, name) if key in self._loaded_cache: return self._loaded_cache[key] - descriptor = matches[0] - if descriptor.api_version not in self.supported_api_versions: - raise ExtensionCompatibilityError(descriptor, tuple(sorted(self.supported_api_versions))) - try: - value = self._load_descriptor(descriptor) - except Exception as exc: # isolate ordinary third-party import failures - raise ExtensionLoadError(descriptor, exc) from exc - with self._lock: + descriptor = matches[0] + if descriptor.api_version not in self.supported_api_versions: + raise ExtensionCompatibilityError(descriptor, tuple(sorted(self.supported_api_versions))) + try: + value = self._load_descriptor(descriptor) + except Exception as exc: # isolate ordinary third-party import failures + raise ExtensionLoadError(descriptor, exc) from exc self._loaded_cache[key] = value - return value + return value def load_all(self, group: str) -> tuple[ExtensionLoadResult, ...]: """Load every allowed extension independently, preserving good results.""" @@ -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, OSError, KeyError, zipfile.BadZipFile) 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: diff --git a/tests/test_extensions.py b/tests/test_extensions.py index a8c7b9e..7fa2b00 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -3,6 +3,7 @@ import importlib import sys import tempfile +import threading import unittest from pathlib import Path from types import SimpleNamespace @@ -76,6 +77,76 @@ 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_refresh_cannot_race_an_inflight_descriptor_load(self) -> None: + entry_point = _entry_point("shared", "same.module:plugin", distribution="trusted") + entered = threading.Event() + release = threading.Event() + + def load() -> str: + entered.set() + release.wait(timeout=5) + return "trusted" + + entry_point.load = load + discovery = base_cli.ExtensionDiscovery(entry_points=(entry_point,), allowlist={"trusted"}) + result: list[object] = [] + + def worker() -> None: + result.append(discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "shared")) + + thread = threading.Thread(target=worker) + thread.start() + self.assertTrue(entered.wait(timeout=5)) + refresh_done = threading.Event() + + def refresh() -> None: + discovery.refresh() + refresh_done.set() + + refresh_thread = threading.Thread(target=refresh) + refresh_thread.start() + self.assertFalse(refresh_done.wait(timeout=0.05)) + release.set() + thread.join(timeout=5) + refresh_thread.join(timeout=5) + self.assertEqual(result, ["trusted"]) + self.assertTrue(refresh_done.is_set()) + def test_malformed_metadata_is_skipped_without_hiding_healthy_extensions(self) -> None: malformed = _entry_point( "broken",