ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ADR-013 Phase 2: meta_path import guard (re-review fix)

Browse Source
This commit is contained in:
Cameron Proudlock
2026-07-18 21:47:40 -04:00
parent 77ca2e0d2d
commit 83a5ce79d9
4 changed files with 368 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
shopdb/plugins/__init__.py
View File

@@ -80,6 +80,16 @@ class PluginManager:
self.loader.verifier = verifier
self.migration_manager.verifier = verifier
# Install (or clear) the import guard that verifies every `plugins.*`
# import against a trusted signature. This is what covers the core
# request handlers that do `from plugins.<name>.models import ...`, which
# never pass through the loader (findings #1, #2).
from . import importguard
if verifier.require_signed:
importguard.install(plugins_dir, verifier)
else:
importguard.uninstall()
# Load enabled plugins
self._load_enabled_plugins()

167
shopdb/plugins/importguard.py Normal file
View File

@@ -0,0 +1,167 @@
"""Import-time verification for ALL plugin code (ADR-013 Phase 2 hardening).
`plugins` is a normal importable package: core request handlers do
`from plugins.<name>.models import ...` through the standard import system,
which never passes through the plugin loader. Gating only load_plugin_class
(plugin.py) therefore left every submodule import unverified - a planted
plugins/<name>/models/*.py executed on an ordinary HTTP request, and a planted
__pycache__/*.pyc ran from a cached read. There is no "single choke point" in
the loader; the choke point is the import system itself.
This installs a sys.meta_path finder that intercepts every `plugins.<name>.*`
import, verifies the plugin's signed provenance once, then verifies each module
file against that provenance and EXECUTES THE EXACT BYTES IT HASHED (read once,
compile, exec) - never a .pyc, never a re-opened file. That closes the submodule
bypass (#1/#2) and the verify-vs-exec TOCTOU (#3) for the import path together.
Installed only under enforcement (PLUGIN_REQUIRE_SIGNED). When off, the finder
is absent and imports behave exactly as before.
"""
import hashlib
import importlib.abc
import importlib.util
import json
import sys
from pathlib import Path
from . import signing
class PluginVerificationError(ImportError):
"""Raised when a plugins.* module is not covered by a trusted signature."""
class _VerifiedSourceLoader(importlib.abc.Loader):
"""Execs source bytes that were already hash-verified (no re-open, no .pyc)."""
def __init__(self, filepath, source_bytes, is_package, search_locations):
self._filepath = str(filepath)
self._source = source_bytes
self._is_package = is_package
self._search = search_locations
def create_module(self, spec):
return None # default module creation
def exec_module(self, module):
code = compile(self._source, self._filepath, 'exec')
exec(code, module.__dict__)
def is_package(self, fullname):
return self._is_package
class PluginImportGuard(importlib.abc.MetaPathFinder):
"""Verifies and loads every plugins.<name>.* module from signed bytes."""
def __init__(self, plugins_dir, verifier):
self.plugins_dir = Path(plugins_dir)
self.verifier = verifier
self._filemaps = {} # plugin name -> verified {relpath: sha256}
def _filemap(self, name):
"""Verify the plugin's provenance signature ONCE, cache its file map."""
if name in self._filemaps:
return self._filemaps[name]
plugin_dir = self.plugins_dir / name
provenance_path = plugin_dir / signing.PROVENANCE_NAME
signature_path = plugin_dir / signing.PROVENANCE_SIG
if not provenance_path.exists() or not signature_path.exists():
raise PluginVerificationError(
f'plugin {name} has no provenance; refusing import under '
f'enforcement')
provenance_bytes = provenance_path.read_bytes()
signature = signature_path.read_bytes()
if not self.verifier._keys or not signing.verify(
self.verifier._keys, provenance_bytes, signature):
raise PluginVerificationError(
f'plugin {name} provenance signature is not trusted')
filemap = json.loads(provenance_bytes).get('files', {})
self._filemaps[name] = filemap
return filemap
def _module_file(self, name, fullname):
"""(filepath, is_package, search_locations) for a plugins.<name>.* module,
or (None, ...) when it is not a source module we should load."""
plugin_dir = self.plugins_dir / name
tail = fullname.split('.')[2:] # components after plugins.<name>
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
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
if fullname == 'plugins':
return None
name = fullname.split('.')[1]
plugin_dir = self.plugins_dir / name
# Dev/external-repo plugins are exempt (only ever under DEBUG/TESTING).
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.
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')
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
def verified_source(self, name, relpath):
"""Return hash-verified bytes of one plugin file (read once), for callers
that load a file explicitly (load_plugin_class + plugin.py). Raises on
any mismatch. Closes the TOCTOU on that file: the caller execs exactly
these bytes."""
filemap = self._filemap(name)
source = (self.plugins_dir / name / relpath).read_bytes()
expected = filemap.get(relpath)
if expected is None or hashlib.sha256(source).hexdigest() != expected:
raise PluginVerificationError(
f'plugin {name} file {relpath} is not covered by a trusted '
f'signature')
return source
def get_installed():
"""Return the installed guard, or None."""
for finder in sys.meta_path:
if isinstance(finder, PluginImportGuard):
return finder
return None
def install(plugins_dir, verifier):
"""Install the guard at the FRONT of sys.meta_path (idempotent, replaces any
prior guard so a re-init picks up new config)."""
uninstall()
guard = PluginImportGuard(plugins_dir, verifier)
sys.meta_path.insert(0, guard)
return guard
def uninstall():
"""Remove any installed guard (used on teardown / when enforcement is off)."""
sys.meta_path[:] = [
f for f in sys.meta_path if not isinstance(f, PluginImportGuard)]

53
shopdb/plugins/loader.py
View File

@@ -160,35 +160,50 @@ class PluginLoader:
plugin_name=name,
)
# THE import gate. Every import path reaches here, so an unsigned or
# tampered plugin never executes under enforcement (findings #1, #2).
# When enforcing: strip any bytecode (so only verified source can run),
# verify the tree, and import without writing new bytecode.
import contextlib
from .packaging import strip_bytecode, no_bytecode
# Under enforcement, verify plugin.py from a SINGLE read and exec exactly
# those bytes (no re-open, no .pyc), so there is no verify-vs-exec gap on
# this file. Its submodule imports (from .models import ...) and every
# `from plugins.X...` in core code are verified by the sys.meta_path
# guard installed at init_app - the import system, not this method, is
# the real choke point (findings #1, #2, #3).
enforcing = self.verifier is not None and self.verifier.require_signed
if enforcing:
from .packaging import strip_bytecode, no_bytecode
from .importguard import PluginImportGuard, PluginVerificationError
strip_bytecode(plugin_dir)
if self.verifier is not None:
ok, reason = self.verifier.check(name)
if not ok:
try:
source = PluginImportGuard(
self.plugins_dir, self.verifier).verified_source(
name, 'plugin.py')
except PluginVerificationError as e:
raise PluginContractError(
f'Plugin {name} failed signature verification: {reason}',
f'Plugin {name} failed signature verification: {e}',
plugin_name=name,
)
try:
with (no_bytecode() if enforcing else contextlib.nullcontext()):
) from e
try:
spec = importlib.util.spec_from_file_location(
f'plugins.{name}.plugin', plugin_module_path)
module = importlib.util.module_from_spec(spec)
with no_bytecode():
exec(compile(source, str(plugin_module_path), 'exec'),
module.__dict__)
except Exception as e:
raise PluginContractError(
f'Plugin {name} import failed: {e}',
plugin_name=name,
) from e
else:
try:
spec = importlib.util.spec_from_file_location(
f'plugins.{name}.plugin', plugin_module_path,
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
except Exception as e:
raise PluginContractError(
f'Plugin {name} import failed: {e}',
plugin_name=name,
) from e
except Exception as e:
raise PluginContractError(
f'Plugin {name} import failed: {e}',
plugin_name=name,
) from e
for attr_name in dir(module):
attr = getattr(module, attr_name)

157
tests/test_plugin_importguard.py Normal file
View File

@@ -0,0 +1,157 @@
"""Import-guard tests (ADR-013 Phase 2 re-review fix).
The critical re-review finding: `plugins` is a normal package, so core code's
`from plugins.<name>.models import ...` bypassed the loader entirely and ran
unverified. These cover the sys.meta_path guard that verifies EVERY plugins.*
import from the exact bytes it execs.
"""
import importlib
import json
import sys
import pytest
from shopdb.plugins import signing
from shopdb.plugins.importguard import (PluginImportGuard,
PluginVerificationError,
install, uninstall, get_installed)
from shopdb.plugins.verification import PluginVerifier
def _write_pkg_plugin(plugins_dir, name, marker='source'):
"""A plugin with a models submodule that records which bytes executed."""
pdir = plugins_dir / name
(pdir / 'models').mkdir(parents=True)
(pdir / 'manifest.json').write_text(json.dumps({
'name': name, 'version': '1.0.0', 'description': 'd',
'core_version': '>=0.1.0,<1.0.0', 'api_prefix': f'/api/{name}',
}))
(pdir / '__init__.py').write_text('')
(pdir / 'plugin.py').write_text('# plugin\n')
(pdir / 'models' / '__init__.py').write_text(f'MARKER = {marker!r}\n')
return pdir
def _stamp(pdir, private_key, name):
provenance = signing.build_provenance(pdir, name, '1.0.0', 'pub')
provenance_bytes = signing.serialize_provenance(provenance)
(pdir / signing.PROVENANCE_NAME).write_bytes(provenance_bytes)
(pdir / signing.PROVENANCE_SIG).write_bytes(
signing.sign(private_key, provenance_bytes))
@pytest.fixture
def keypair(tmp_path):
private_pem, public_pem = signing.generate_keypair()
pub_path = tmp_path / 'k.pub'
pub_path.write_bytes(public_pem)
return signing.load_private_key(private_pem), str(pub_path)
def _verifier(plugins_dir, pub_path):
return PluginVerifier(plugins_dir, require_signed=True,
trusted_key_paths=[pub_path])
@pytest.fixture
def guarded(tmp_path, keypair):
"""Install a guard, yield (guard, plugins_dir, private_key, name), clean up
sys.modules + sys.meta_path afterward so tests do not leak."""
private_key, pub_path = keypair
plugins_dir = tmp_path / 'plugins'
plugins_dir.mkdir()
name = 'guardtestpkg'
guard = install(plugins_dir, _verifier(plugins_dir, pub_path))
try:
yield guard, plugins_dir, private_key, name
finally:
uninstall()
for mod in list(sys.modules):
if mod == f'plugins.{name}' or mod.startswith(f'plugins.{name}.'):
del sys.modules[mod]
# --- direct find_spec / verified_source -------------------------------------
def test_verified_source_signed_ok_tampered_raises(guarded):
guard, plugins_dir, priv, name = guarded
pdir = _write_pkg_plugin(plugins_dir, name)
_stamp(pdir, priv, name)
assert guard.verified_source(name, 'plugin.py') == b'# plugin\n'
(pdir / 'plugin.py').write_text('# tampered\n')
with pytest.raises(PluginVerificationError):
guard.verified_source(name, 'plugin.py')
def test_find_spec_refuses_unsigned_plugin(guarded):
guard, plugins_dir, priv, name = guarded
_write_pkg_plugin(plugins_dir, name) # no provenance stamped
with pytest.raises(PluginVerificationError):
guard.find_spec(f'plugins.{name}.models')
def test_find_spec_refuses_tampered_submodule(guarded):
guard, plugins_dir, priv, name = guarded
pdir = _write_pkg_plugin(plugins_dir, name)
_stamp(pdir, priv, name)
# tamper the submodule after signing
(pdir / 'models' / '__init__.py').write_text('MARKER = "evil"\n')
with pytest.raises(PluginVerificationError):
guard.find_spec(f'plugins.{name}.models')
# --- real import through the installed guard --------------------------------
def test_core_style_submodule_import_is_verified(guarded):
"""`from plugins.<name>.models import MARKER` runs the SIGNED source."""
guard, plugins_dir, priv, name = guarded
pdir = _write_pkg_plugin(plugins_dir, name, marker='source')
_stamp(pdir, priv, name)
module = importlib.import_module(f'plugins.{name}.models')
assert module.MARKER == 'source'
def test_planted_pyc_is_never_consulted(guarded):
"""A planted __pycache__/*.pyc must not run: the guard loads verified source
only, so a garbage bytecode cache is simply ignored (finding #1/#2)."""
guard, plugins_dir, priv, name = guarded
pdir = _write_pkg_plugin(plugins_dir, name, marker='source')
_stamp(pdir, priv, name)
cache = pdir / 'models' / '__pycache__'
cache.mkdir()
# garbage bytecode: if the loader ever honored it, import would crash
for pyc in (f'__init__.cpython-{sys.version_info.major}'
f'{sys.version_info.minor}.pyc',):
(cache / pyc).write_bytes(b'\x00\x00\x00\x00garbage-not-valid-bytecode')
module = importlib.import_module(f'plugins.{name}.models')
assert module.MARKER == 'source' # source ran, pyc ignored
def test_tampered_submodule_import_raises(guarded):
guard, plugins_dir, priv, name = guarded
pdir = _write_pkg_plugin(plugins_dir, name)
_stamp(pdir, priv, name)
(pdir / 'models' / '__init__.py').write_text('MARKER = "evil"\n')
with pytest.raises(ImportError):
importlib.import_module(f'plugins.{name}.models')
def test_install_uninstall_idempotent(tmp_path, keypair):
_, pub_path = keypair
plugins_dir = tmp_path / 'plugins'
plugins_dir.mkdir()
verifier = _verifier(plugins_dir, pub_path)
try:
install(plugins_dir, verifier)
install(plugins_dir, verifier) # replaces, does not stack
assert sum(isinstance(f, PluginImportGuard) for f in sys.meta_path) == 1
assert get_installed() is not None
finally:
uninstall()
assert get_installed() is None