diff --git a/docs/PLUGIN-SIGNING.md b/docs/PLUGIN-SIGNING.md index 0a7b6fd..bcfd9c3 100644 --- a/docs/PLUGIN-SIGNING.md +++ b/docs/PLUGIN-SIGNING.md @@ -88,12 +88,27 @@ signatures on a site: PLUGIN_REQUIRE_SIGNED=true ``` -Now a plugin only loads (verify-at-load) or migrates (verify-at-migrate) when -its tree matches a trusted signature. An unsigned, tampered, or wrong-key -plugin is refused - fail-closed. `PLUGIN_DEV_TRUST_DIRS` exempts named +Now a plugin only loads or migrates when its tree matches a trusted signature. +Under enforcement a `sys.meta_path` guard verifies EVERY `plugins..*` +import (not just `plugin.py`) - including the `from plugins..models import +...` that core request handlers do - against the plugin's signed provenance, and +executes the exact bytes it hashed (never a `.pyc`). An unsigned, tampered, or +wrong-key plugin is refused, fail-closed. `PLUGIN_DEV_TRUST_DIRS` exempts named directories, but ONLY under DEBUG/TESTING (the external-repo dev workflow); production ignores it. +Because every `plugins.*` import is verified, `stamp-bundled` must cover EVERY +plugin directory present (its no-argument form does), not only the enabled ones +- core code can import a disabled plugin's module, and an unstamped one would be +refused. + +Defense in depth - set filesystem permissions so the app's runtime user CANNOT +write the `plugins/` directory (owned by the deploy user). Import-time +verification closes the "attacker drops a file, a request imports it" path; a +strict read-only `plugins/` also closes the narrow verify-vs-migrate race where +an attacker with concurrent write to `plugins/` swaps a migration script between +the check and alembic re-reading it. + ## The shelf and adopt (Phase 2) A shelf is a read-only folder of artifacts plus a signed index. The app reads diff --git a/shopdb/plugins/__init__.py b/shopdb/plugins/__init__.py index 5e7f06d..64ad849 100644 --- a/shopdb/plugins/__init__.py +++ b/shopdb/plugins/__init__.py @@ -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..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() diff --git a/shopdb/plugins/importguard.py b/shopdb/plugins/importguard.py new file mode 100644 index 0000000..32a84fc --- /dev/null +++ b/shopdb/plugins/importguard.py @@ -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..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//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..*` +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..* 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..* 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. + 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)] diff --git a/shopdb/plugins/loader.py b/shopdb/plugins/loader.py index d732883..6264ac7 100644 --- a/shopdb/plugins/loader.py +++ b/shopdb/plugins/loader.py @@ -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) diff --git a/tests/test_plugin_importguard.py b/tests/test_plugin_importguard.py new file mode 100644 index 0000000..928f8e1 --- /dev/null +++ b/tests/test_plugin_importguard.py @@ -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..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..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