feat(packs): let an operator decline to execute in-pack provider code

Closes the three items in #2263.

1. AUTHORITY_PACK_LOAD_PROVIDERS (default True, so nothing changes for
   existing installs). Loading an in-pack provider IMPORTS it into the web and
   worker processes; extraction refuses path traversal and setuid bits but
   cannot refuse code. When off, the modules are NOT imported and the skip is
   logged once with a count and the module names, so 'turned off' and 'no packs
   installed' stay distinguishable. Safe to decline because the pack contract
   (authority-packs SOURCE_PROVIDERS.md, P5) requires a pack to serve its text
   with providers/ deleted.

2. get_authority_source_provider(class_name) — the supported seam for an
   in-pack provider that DELEGATES to a core one rather than re-scraping.
   Accepts a bare class name or a dotted path, refuses an ambiguous leaf rather
   than picking, and returns None (never raises) so a pack declines instead of
   crashing registry build.

3. install_authority_pack now prints the provider modules a pack ships BEFORE
   any DB writes — by listing files and reading the optional providers: block,
   never by importing them, since reporting a pack's code by executing it would
   defeat the point.

test_authority_pack_provider_trust.py probes with a module that writes a
sentinel AT IMPORT, so 'was it executed?' is answerable independently of
whether its class registered. Verified the two gate tests fail with the gate
removed and pass with it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: JSv4

config/settings/base.py

@@ -1784,6 +1784,22 @@
 # path. Same restart requirement as AUTHORITY_PACK_PATHS.
 AUTHORITY_PACK_ROOTS = env.list("AUTHORITY_PACK_ROOTS", default=[])
 
+# Whether to LOAD (import) provider modules shipped inside authority packs
+# (``<pack>/providers/*.py``, ``<pack>/discovery_providers/*.py``).
+#
+# Installing such a pack executes its Python in the web and worker processes.
+# That is a materially larger blast radius than ``source_hosts``, where
+# "installing the pack is the trust decision" holds because the consequence is
+# bounded to which hosts may be fetched. Extraction already refuses path
+# traversal and setuid bits (``tar.extract(..., filter="data")``); it cannot
+# refuse code.
+#
+# Default True preserves existing behaviour. An operator installing packs they
+# did not author sets this False and loses only the ability to RE-FETCH the
+# pack's text: the authority-packs contract (SOURCE_PROVIDERS.md, clause P5)
+# requires a pack to install and serve its sections with ``providers/`` deleted.
+AUTHORITY_PACK_LOAD_PROVIDERS = env.bool("AUTHORITY_PACK_LOAD_PROVIDERS", default=True)
+
 # Where `manage.py install_authority_pack` materialises packs fetched from the
 # pack registry repo. The directory is an implicit pack bundle root (scanned by
 # authority_pack_dirs() exactly like an AUTHORITY_PACK_ROOTS entry), so a

opencontractserver/corpuses/management/commands/install_authority_pack.py

@@ -140,6 +140,63 @@ def materialise_pack(staged_pack: Path, pack: str, stdout=None) -> Path:
     return dest
 
 
+def _report_pack_providers(pack_dir, stdout, style) -> None:
+    """Print the provider modules a pack ships, if any. Never imports them.
+
+    Installing a pack that ships ``providers/`` imports its Python into the web
+    and worker processes, so the operator should be able to see that surface
+    before the install writes anything. Reads the filesystem and, when present,
+    the OPTIONAL ``providers:`` declaration in pack.yaml — never the modules.
+    """
+    from pathlib import Path as _Path
+
+    pack_dir = _Path(pack_dir)
+    shipped: list[tuple[str, str]] = []
+    for subdir in ("providers", "discovery_providers"):
+        component_dir = pack_dir / subdir
+        if not component_dir.is_dir():
+            continue
+        for py in sorted(component_dir.glob("*.py")):
+            if not py.name.startswith("_"):
+                shipped.append((subdir, py.name))
+    if not shipped:
+        return
+
+    stdout.write(
+        style.WARNING(
+            f"This pack ships {len(shipped)} provider module(s). Loading the pack "
+            "IMPORTS them into the web and worker processes:"
+        )
+    )
+    for subdir, name in shipped:
+        stdout.write(f"    {subdir}/{name}")
+
+    # The declaration is optional and descriptive; show it when a pack has one
+    # so the claimed prefixes are visible without reading Python.
+    try:
+        import yaml
+
+        manifest = yaml.safe_load((pack_dir / "pack.yaml").read_text()) or {}
+        for entry in manifest.get("providers") or []:
+            stdout.write(
+                f"    declares: {entry.get('class')} "
+                f"prefixes={entry.get('supported_prefixes')} "
+                f"delegates_to={entry.get('delegates_to') or '-'}"
+            )
+    except Exception:  # noqa: BLE001 - a missing/!parsing manifest is not fatal here
+        pass
+
+    from django.conf import settings
+
+    if not getattr(settings, "AUTHORITY_PACK_LOAD_PROVIDERS", True):
+        stdout.write(
+            style.SUCCESS(
+                "    AUTHORITY_PACK_LOAD_PROVIDERS is off — these will NOT be "
+                "imported. The pack's text still installs and serves."
+            )
+        )
+
+
 class Command(BaseCommand):
     help = (
         "Fetch an authority pack from the pack registry repo into "
@@ -282,6 +339,11 @@ def handle(self, *args, **options):
                 "--creator is required to install or preflight (or use --fetch-only)"
             )
 
+        # Say what code this pack will run, BEFORE any DB writes, and without
+        # importing it — reporting a pack's providers by executing them would
+        # defeat the point. Static file listing only.
+        _report_pack_providers(dest, self.stdout, self.style)
+
         call_command(
             "load_authority_pack",
             path=str(dest),

opencontractserver/pipeline/registry.py

@@ -228,6 +228,27 @@ def authority_pack_dirs() -> list[Path]:
     return unique
 
 
+def pack_provider_modules(subdir_name: str) -> list[Path]:
+    """Provider modules a pack SHIPS, found without importing any of them.
+
+    Used for two things that must not execute pack code: telling an operator
+    what a pack will run before they install it, and naming what was skipped
+    when loading is switched off. Mirrors the filtering
+    ``_discover_pack_component_classes`` applies (``*.py``, no leading
+    underscore) so the count is the count that matters.
+    """
+    modules: list[Path] = []
+    for pack_dir in authority_pack_dirs():
+        component_dir = pack_dir / subdir_name
+        if not component_dir.is_dir():
+            continue
+        modules.extend(
+            py for py in sorted(component_dir.glob("*.py"))
+            if not py.name.startswith("_")
+        )
+    return modules
+
+
 class ComponentType(str, Enum):
     """Types of pipeline components."""
 
@@ -472,6 +493,26 @@ def _discover_pack_component_classes(
         classes that merely happen to be in scope are ignored via the
         ``__module__`` check.
         """
+        # Loading an in-pack provider IMPORTS it into this process. That is the
+        # trust decision AUTHORITY_PACK_LOAD_PROVIDERS exists to let an operator
+        # decline; see the setting's comment in config/settings/base.py.
+        # Reported rather than silent, and reported with a count, so "I turned
+        # it off" and "this install has none" are distinguishable.
+        from django.conf import settings as _settings
+
+        if not getattr(_settings, "AUTHORITY_PACK_LOAD_PROVIDERS", True):
+            skipped = pack_provider_modules(subdir_name)
+            if skipped:
+                logger.info(
+                    "AUTHORITY_PACK_LOAD_PROVIDERS is off: skipped %d in-pack "
+                    "%s module(s) without importing them (%s). The packs still "
+                    "serve their text; only re-fetch is unavailable.",
+                    len(skipped),
+                    subdir_name,
+                    ", ".join(sorted(p.parent.parent.name + "/" + p.name for p in skipped)),
+                )
+            return []
+
         seen: set[type] = set()
         found: list[type] = []
         pack_dirs = authority_pack_dirs()
@@ -1020,6 +1061,66 @@ def get_all_authority_source_providers_cached() -> (
     return get_registry().authority_source_providers
 
 
+def get_authority_source_provider(class_name: str) -> Optional[Any]:
+    """Return an INSTANCE of a registered authority source provider, by class name.
+
+    The supported seam for an in-pack provider that DELEGATES to a core one.
+
+    Most packs should not ship a scraper: core already covers the Code of
+    Federal Regulations, the U.S. Code and the Federal Register, and those
+    providers fail to fire for a pack only because of key SHAPE — the CFR
+    provider accepts ``cfr-{digits}:`` while a pack's sections are keyed
+    ``itar:``, ``ear:``, ``aeca:``. A pack already declares that translation in
+    its own equivalence rows, so the in-pack provider is a key translator plus a
+    delegation (authority-packs ``SOURCE_PROVIDERS.md``).
+
+    Doing that by importing the core class directly works, but couples every
+    such pack to an import path: a refactor of the core providers would break
+    them, and registry discovery isolates import failures by design, so the
+    breakage would be a logged warning and a provider that silently stopped
+    existing. Depending on this function instead makes the class name the
+    contract.
+
+    Returns ``None`` when no provider of that class name is registered — a pack
+    MUST handle that by declining (``can_handle`` -> False) rather than raising,
+    so an install missing a core provider degrades to "cannot re-fetch" instead
+    of breaking registry build.
+    """
+    registry = get_registry()
+    definition = registry.get_by_class_name(class_name) or registry.get_by_name(class_name)
+    if definition is None:
+        # ``class_name`` is stored as a full dotted path; a pack author writes
+        # the leaf. Accept either, and refuse an ambiguous leaf rather than
+        # picking — two providers with the same class name in different
+        # packages is exactly when silently choosing is worst.
+        matches = [
+            d
+            for d in get_all_authority_source_providers_cached()
+            if d.class_name.rsplit(".", 1)[-1] == class_name
+        ]
+        if len(matches) > 1:
+            logger.warning(
+                "Authority source provider name %r is ambiguous (%s); "
+                "delegate using the full dotted path.",
+                class_name,
+                ", ".join(sorted(d.class_name for d in matches)),
+            )
+            return None
+        if not matches:
+            return None
+        definition = matches[0]
+    component = getattr(definition, "component_class", None)
+    if component is None:
+        return None
+    try:
+        return component()
+    except Exception as exc:  # pragma: no cover - defensive; a provider ctor should be trivial
+        logger.warning(
+            "Could not instantiate authority source provider %r: %s", class_name, exc
+        )
+        return None
+
+
 def get_all_authority_discovery_providers_cached() -> (
     tuple[PipelineComponentDefinition, ...]
 ):

opencontractserver/tests/test_authority_pack_provider_trust.py

@@ -0,0 +1,175 @@
+"""Declining to execute a pack's code, and the supported way to reuse a core provider.
+
+Installing an authority pack that ships ``<pack>/providers/*.py`` IMPORTS that
+Python into the web and worker processes. Extraction already refuses path
+traversal and setuid bits (``tar.extract(..., filter="data")``); it cannot
+refuse code. That is a materially larger blast radius than ``source_hosts``,
+where "installing the pack is the trust decision" holds because the consequence
+is bounded to which hosts may be fetched.
+
+``AUTHORITY_PACK_LOAD_PROVIDERS`` lets an operator decline. It is safe to
+decline because the pack contract (authority-packs ``SOURCE_PROVIDERS.md``,
+clause P5) requires a pack to install and serve its sections with ``providers/``
+deleted — so turning it off costs re-fetch and nothing else.
+
+The tests below use a provider module that writes a sentinel **at import time**.
+That is exactly what a pack MUST NOT do (clause P4), which is what makes it a
+good probe here: if the sentinel appears, the module was imported, and no
+assertion about registration can be fooled by a provider that merely failed to
+register.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+from django.test import SimpleTestCase, override_settings
+
+from opencontractserver.pipeline.registry import (
+    get_all_authority_source_providers_cached,
+    get_authority_source_provider,
+    pack_provider_modules,
+    reset_registry,
+)
+
+_REGISTRY_LOGGER = "opencontractserver.pipeline.registry"
+
+# Writes a file when imported, so "was this module executed?" is answerable
+# independently of whether its class ended up registered.
+_PROVIDER_SRC = '''
+from pathlib import Path
+
+from opencontractserver.pipeline.base.base_authority_source_provider import (
+    AuthorityRequest,
+    BaseAuthoritySourceProvider,
+)
+
+Path(__file__).with_name("IMPORTED.sentinel").write_text("yes")
+
+
+class TrustProbeSourceProvider(BaseAuthoritySourceProvider):
+    title = "Trust Probe"
+    description = "Fixture provider for AUTHORITY_PACK_LOAD_PROVIDERS tests."
+    author = "test"
+    supported_prefixes = ("trustprobe",)
+
+    def can_handle(self, canonical_key: str) -> bool:
+        return canonical_key.split(":", 1)[0] == "trustprobe"
+
+    def _locate_impl(self, canonical_key, **kwargs):
+        return AuthorityRequest(url="https://example.invalid/x")
+
+    def _fetch_impl(self, request, **kwargs):
+        raise NotImplementedError
+'''
+
+
+class _PackFixture(SimpleTestCase):
+    """Builds a throwaway pack dir containing one provider module."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.pack_dir = Path(self._tmp.name) / "trustprobe_pack"
+        providers = self.pack_dir / "providers"
+        providers.mkdir(parents=True)
+        (providers / "trust_probe_provider.py").write_text(_PROVIDER_SRC)
+        self.sentinel = providers / "IMPORTED.sentinel"
+        reset_registry()
+        self.addCleanup(reset_registry)
+
+    def _class_names(self) -> set[str]:
+        # ``class_name`` is the full dotted path; compare on the leaf so the
+        # assertions read as the class a pack author actually writes.
+        return {
+            d.class_name.rsplit(".", 1)[-1]
+            for d in get_all_authority_source_providers_cached()
+        }
+
+
+class LoadProvidersSettingTests(_PackFixture):
+    def test_providers_load_by_default(self) -> None:
+        """The default must not change: existing installs keep working."""
+        with override_settings(
+            AUTHORITY_PACK_PATHS=[str(self.pack_dir)],
+            AUTHORITY_PACK_LOAD_PROVIDERS=True,
+        ):
+            names = self._class_names()
+
+        self.assertIn("TrustProbeSourceProvider", names)
+        self.assertTrue(
+            self.sentinel.exists(), "fixture never imported; the test proves nothing"
+        )
+
+    def test_providers_are_not_imported_when_disabled(self) -> None:
+        """Not merely unregistered — NOT EXECUTED."""
+        with override_settings(
+            AUTHORITY_PACK_PATHS=[str(self.pack_dir)],
+            AUTHORITY_PACK_LOAD_PROVIDERS=False,
+        ):
+            names = self._class_names()
+
+        self.assertNotIn("TrustProbeSourceProvider", names)
+        self.assertFalse(
+            self.sentinel.exists(),
+            "the pack's module was IMPORTED despite AUTHORITY_PACK_LOAD_PROVIDERS "
+            "being off — the setting gates registration but not execution, which "
+            "is the only thing it was for",
+        )
+
+    def test_skipping_is_reported_with_a_count(self) -> None:
+        """Silence would make 'turned off' and 'no packs installed' identical."""
+        with override_settings(
+            AUTHORITY_PACK_PATHS=[str(self.pack_dir)],
+            AUTHORITY_PACK_LOAD_PROVIDERS=False,
+        ):
+            with self.assertLogs(_REGISTRY_LOGGER, level="INFO") as captured:
+                self._class_names()
+
+        messages = "\n".join(captured.output)
+        self.assertIn("AUTHORITY_PACK_LOAD_PROVIDERS is off", messages)
+        self.assertIn("trust_probe_provider.py", messages)
+
+    def test_core_providers_survive_the_setting(self) -> None:
+        """Only IN-PACK loading is declined; core providers are unaffected."""
+        with override_settings(
+            AUTHORITY_PACK_PATHS=[str(self.pack_dir)],
+            AUTHORITY_PACK_LOAD_PROVIDERS=False,
+        ):
+            names = self._class_names()
+
+        self.assertIn("CFRAuthoritySourceProvider", names)
+
+
+class PackProviderModulesTests(_PackFixture):
+    def test_lists_shipped_modules_without_importing_them(self) -> None:
+        """The listing that --check and the skip log both depend on."""
+        with override_settings(AUTHORITY_PACK_PATHS=[str(self.pack_dir)]):
+            found = pack_provider_modules("providers")
+
+        self.assertEqual([p.name for p in found], ["trust_probe_provider.py"])
+        self.assertFalse(
+            self.sentinel.exists(), "listing the modules executed one of them"
+        )
+
+
+class DelegationSeamTests(SimpleTestCase):
+    """``get_authority_source_provider`` is the seam in-pack providers delegate through."""
+
+    def setUp(self) -> None:
+        reset_registry()
+        self.addCleanup(reset_registry)
+
+    def test_returns_a_usable_core_provider_instance(self) -> None:
+        provider = get_authority_source_provider("CFRAuthoritySourceProvider")
+
+        self.assertIsNotNone(provider)
+        # The thing a delegating pack provider actually needs: routing plus a
+        # pure locate it can hand a translated key to.
+        self.assertTrue(provider.can_handle("cfr-22:120.4"))
+        self.assertFalse(provider.can_handle("itar:120.4"))
+
+    def test_unknown_provider_returns_none_rather_than_raising(self) -> None:
+        """A pack must be able to decline, not crash registry build."""
+        self.assertIsNone(get_authority_source_provider("NoSuchProviderClass"))