From c59d2dab562bc7a03a24879ab10e38ef452686e7 Mon Sep 17 00:00:00 2001 From: cproudlock Date: Sat, 18 Jul 2026 22:12:20 -0400 Subject: [PATCH] ADR-013 Phase 2: import guard fails closed on non-.py + owns package root Third review found the meta_path guard leaked exactly where it delegated to the stdlib import system: 1. Non-.py submodules (CRITICAL). When a name had no dir and no .py, find_spec returned None and the stdlib loaded a planted .so (ExtensionFileLoader) or a sourceless .pyc unverified - an attacker deletes a signed .py and drops a same-named .so with arbitrary init code, run on a normal request via core's `from plugins..models import ...`. The guard now refuses any name for which a non-source importable candidate (EXTENSION_SUFFIXES + BYTECODE_ SUFFIXES) exists on disk; None is reserved for genuinely-absent modules. 2. Top-level plugins/__init__.py (CRITICAL). It is in no plugin's provenance, is attacker-writable, and Python runs it before any guarded submodule. The guard now owns `plugins`: it execs an EMPTY package body (search points at the plugins dir), so an overwritten plugins/__init__.py never runs. Also: specs are built with spec_from_file_location so loaded modules get __file__/__path__ (Flask blueprint root paths need it) while the loader still execs the verified in-memory bytes - never re-reading the file. Verified end to end: under PLUGIN_REQUIRE_SIGNED with all 13 bundled plugins stamped, the app boots and loads every plugin through the guard; a tampered plugin file is refused at load. 4 new guard tests (planted .so, sourceless .pyc, absent-module defer, neutralized package root). Prior fixes #3/#4 confirmed still sound by the review. 1065 pass, naming green. --- shopdb/plugins/importguard.py | 104 ++++++++++++++++++++++--------- tests/test_plugin_importguard.py | 46 ++++++++++++++ 2 files changed, 119 insertions(+), 31 deletions(-) diff --git a/shopdb/plugins/importguard.py b/shopdb/plugins/importguard.py index 32a84fc..ad91b45 100644 --- a/shopdb/plugins/importguard.py +++ b/shopdb/plugins/importguard.py @@ -20,6 +20,7 @@ is absent and imports behave exactly as before. import hashlib import importlib.abc +import importlib.machinery import importlib.util import json import sys @@ -27,6 +28,14 @@ from pathlib import Path from . import signing +# Non-source files the stdlib import system would execute for a module name. +# The guard loads ONLY verified .py source, so any of these under a plugin is a +# module the guard cannot vouch for and must refuse (a planted .so or a +# sourceless .pyc standing in for a deleted, signed .py). +_UNVERIFIABLE_SUFFIXES = tuple( + importlib.machinery.EXTENSION_SUFFIXES + + importlib.machinery.BYTECODE_SUFFIXES) + class PluginVerificationError(ImportError): """Raised when a plugins.* module is not covered by a trusted signature.""" @@ -51,6 +60,14 @@ class _VerifiedSourceLoader(importlib.abc.Loader): def is_package(self, fullname): return self._is_package + def get_filename(self, fullname): + # gives the loaded module a __file__ so Flask can resolve blueprint + # root paths (spec.origin is set from this). + return self._filepath + + def get_source(self, fullname): + return self._source.decode('utf-8', 'replace') + class PluginImportGuard(importlib.abc.MetaPathFinder): """Verifies and loads every plugins..* module from signed bytes.""" @@ -81,25 +98,49 @@ class PluginImportGuard(importlib.abc.MetaPathFinder): self._filemaps[name] = filemap return filemap - def _module_file(self, name, fullname): - """(filepath, is_package, search_locations) for a plugins..* module, - or (None, ...) when it is not a source module we should load.""" + def _refuse_if_unverifiable_present(self, directory, leaf, name): + """Fail closed if a non-.py importable file (.so / sourceless .pyc) for + `leaf` sits in `directory`. Reserving None for genuinely-absent names is + what stops the stdlib loading a planted extension/bytecode unverified.""" + for suffix in _UNVERIFIABLE_SUFFIXES: + if (directory / f'{leaf}{suffix}').exists(): + raise PluginVerificationError( + f'plugin {name} module {leaf}{suffix} is not loadable from ' + f'trusted source; refusing') + + def _source_spec(self, name, fullname, filepath, is_package, search): plugin_dir = self.plugins_dir / name - tail = fullname.split('.')[2:] # components after plugins. - base = plugin_dir.joinpath(*tail) if tail else plugin_dir - if base.is_dir(): - return base / '__init__.py', True, [str(base)] - source = base.with_suffix('.py') - if source.exists(): - return source, False, None - return None, False, None + relpath = filepath.relative_to(plugin_dir).as_posix() + source = filepath.read_bytes() + expected = self._filemap(name).get(relpath) + if expected is None or hashlib.sha256(source).hexdigest() != expected: + raise PluginVerificationError( + f'plugin {name} module {relpath} is not covered by a trusted ' + f'signature') + loader = _VerifiedSourceLoader(filepath, source, is_package, search) + # spec_from_file_location sets origin -> the module gets __file__/__path__, + # which Flask needs for blueprint root paths; the loader still execs the + # verified in-memory bytes, never re-reading the file. + return importlib.util.spec_from_file_location( + fullname, str(filepath), loader=loader, + submodule_search_locations=search if is_package else None) def find_spec(self, fullname, path=None, target=None): - # Only our package; the empty top-level `plugins` package loads normally. if fullname != 'plugins' and not fullname.startswith('plugins.'): return None + + # The top-level `plugins` package is in no plugin's provenance, yet + # Python imports it before any guarded submodule. Own it: execute an + # EMPTY package body so an attacker-overwritten plugins/__init__.py never + # runs, and point package search at the plugins dir. if fullname == 'plugins': - return None + init_path = self.plugins_dir / '__init__.py' + loader = _VerifiedSourceLoader( + init_path, b'', True, [str(self.plugins_dir)]) + return importlib.util.spec_from_file_location( + 'plugins', str(init_path), loader=loader, + submodule_search_locations=[str(self.plugins_dir)]) + name = fullname.split('.')[1] plugin_dir = self.plugins_dir / name @@ -107,27 +148,28 @@ class PluginImportGuard(importlib.abc.MetaPathFinder): if self.verifier._dev_exempt(plugin_dir): return None - filemap = self._filemap(name) # verify signature (raises on failure) - filepath, is_package, search = self._module_file(name, fullname) - if filepath is None or not filepath.exists(): - # A missing __init__.py (namespace pkg) or non-python target: let the - # normal machinery decide. Any .py it would run is covered above. + self._filemap(name) # verify the plugin signature (raises on failure) + + tail = fullname.split('.')[2:] # components after plugins. + base = plugin_dir.joinpath(*tail) if tail else plugin_dir + + if base.is_dir(): + init = base / '__init__.py' + if init.exists(): + return self._source_spec(name, fullname, init, True, [str(base)]) + # Regular package with no __init__.py: refuse a planted + # __init__.; otherwise treat as a namespace package (no body). + self._refuse_if_unverifiable_present(base, '__init__', name) return None - relpath = filepath.relative_to(plugin_dir).as_posix() - source = filepath.read_bytes() - expected = filemap.get(relpath) - if expected is None or hashlib.sha256(source).hexdigest() != expected: - raise PluginVerificationError( - f'plugin {name} module {relpath} is not covered by a trusted ' - f'signature') + source = base.with_suffix('.py') + if source.exists(): + return self._source_spec(name, fullname, source, False, None) - loader = _VerifiedSourceLoader(filepath, source, is_package, search) - spec = importlib.util.spec_from_loader( - fullname, loader, is_package=is_package) - if is_package: - spec.submodule_search_locations = search - return spec + # No .py for this name: refuse a planted .so / sourceless .pyc, else defer + # (a genuinely-absent module -> normal ModuleNotFoundError). + self._refuse_if_unverifiable_present(base.parent, base.name, name) + return None def verified_source(self, name, relpath): """Return hash-verified bytes of one plugin file (read once), for callers diff --git a/tests/test_plugin_importguard.py b/tests/test_plugin_importguard.py index 928f8e1..7914ba7 100644 --- a/tests/test_plugin_importguard.py +++ b/tests/test_plugin_importguard.py @@ -142,6 +142,52 @@ def test_tampered_submodule_import_raises(guarded): importlib.import_module(f'plugins.{name}.models') +def test_planted_extension_module_refused(guarded): + """Finding (round 3): a planted .so standing in for a would-be submodule must + be refused, not handed to the stdlib ExtensionFileLoader unverified.""" + import importlib.machinery + guard, plugins_dir, priv, name = guarded + pdir = _write_pkg_plugin(plugins_dir, name) + _stamp(pdir, priv, name) + + ext = importlib.machinery.EXTENSION_SUFFIXES[0] + (pdir / 'models' / f'evil{ext}').write_bytes(b'\x7fELF-not-really') + with pytest.raises(PluginVerificationError): + guard.find_spec(f'plugins.{name}.models.evil') + + +def test_planted_sourceless_pyc_refused(guarded): + """A sourceless .pyc for a name with no .py must be refused.""" + guard, plugins_dir, priv, name = guarded + pdir = _write_pkg_plugin(plugins_dir, name) + _stamp(pdir, priv, name) + + (pdir / 'models' / 'evil.pyc').write_bytes(b'\x00garbage') + with pytest.raises(PluginVerificationError): + guard.find_spec(f'plugins.{name}.models.evil') + + +def test_absent_module_defers(guarded): + """A genuinely-absent module returns None (normal ModuleNotFoundError).""" + guard, plugins_dir, priv, name = guarded + pdir = _write_pkg_plugin(plugins_dir, name) + _stamp(pdir, priv, name) + assert guard.find_spec(f'plugins.{name}.models.notthere') is None + + +def test_top_level_plugins_body_neutralized(guarded): + """Finding (round 3): the top-level plugins/__init__.py runs first and is in + no provenance; the guard must exec an EMPTY body so an attacker-overwritten + plugins/__init__.py cannot execute.""" + guard, plugins_dir, priv, name = guarded + (plugins_dir / '__init__.py').write_text('raise RuntimeError("attacker code")') + spec = guard.find_spec('plugins') + assert spec is not None + # the loader carries empty source, not the on-disk body + assert spec.loader._source == b'' + assert str(plugins_dir) in spec.submodule_search_locations + + def test_install_uninstall_idempotent(tmp_path, keypair): _, pub_path = keypair plugins_dir = tmp_path / 'plugins'