From 55a6f1b8d3d7c63f6223f9b120ae1984da232321 Mon Sep 17 00:00:00 2001 From: cproudlock Date: Sat, 18 Jul 2026 21:06:27 -0400 Subject: [PATCH] ADR-013 Phase 2: fix four bypasses found by adversarial review An adversarial security review of the Phase 2 trust model found four real bypasses (two remote-triggerable to in-process code execution). Root cause for three: the set of bytes verification covered was smaller than the set that determined execution. Fixes: 1. Bytecode-cache blind spot (CRITICAL). verify_dir excluded __pycache__/.pyc, so a planted cache ran while escaping the hash map. verify_dir now flags any bytecode as an unexpected file; the loader strips bytecode before verify and imports under sys.dont_write_bytecode, so only verified source executes. 2. Unauthenticated verify-at-load bypass (CRITICAL). load_plugin_class imported plugin.py with no gate, reachable via discover_available / an anonymous GET /api/plugins. The verify+strip gate moved INTO load_plugin_class - the single import choke point every path flows through - so an unsigned/tampered plugin is never imported. discover_available skips a refused plugin instead of 500. 3. Ungated migration entrypoints (HIGH). downgrade_plugin and get_current_head (ScriptDirectory imports version modules) ran plugin code with no check. All alembic-invoking methods now pass through _verify_ok (strip + verify) first and run under no-bytecode. 4. Revocation/content bypass (HIGH). The signed index bound a filename, not content; adopt did not bind the delivered bytes to the resolved version, so revoked bytes could be served under a live filename. The index now records a per-artifact SHA-256; adopt verifies the on-disk digest and requires the artifact's own signed manifest version to equal the resolved version. Enforcement stays default-off; strip/no-bytecode run only under enforcement, so the unsigned path is unchanged. 6 regression tests (planted bytecode, the discover import path, downgrade gate, version-swap). 1054 pass, naming green. --- shopdb/plugins/__init__.py | 9 ++++- shopdb/plugins/cli.py | 11 +++++- shopdb/plugins/loader.py | 44 +++++++++++++-------- shopdb/plugins/migrations.py | 66 +++++++++++++++++++++++-------- shopdb/plugins/packaging.py | 44 +++++++++++++++++++-- shopdb/plugins/shelf.py | 44 +++++++++++++++++---- shopdb/plugins/signing.py | 7 +++- tests/test_plugin_packaging.py | 4 ++ tests/test_plugin_shelf.py | 39 +++++++++++++++++- tests/test_plugin_verification.py | 55 ++++++++++++++++++++++++++ 10 files changed, 273 insertions(+), 50 deletions(-) diff --git a/shopdb/plugins/__init__.py b/shopdb/plugins/__init__.py index abe3d32..5e7f06d 100644 --- a/shopdb/plugins/__init__.py +++ b/shopdb/plugins/__init__.py @@ -154,7 +154,14 @@ class PluginManager: available = [] for name in self.loader.discover_plugins(): - plugin_class = self.loader.load_plugin_class(name) + # Under enforcement load_plugin_class refuses an unverified plugin + # (it will not import it). Skip such a plugin from the listing + # instead of failing the whole call. + try: + plugin_class = self.loader.load_plugin_class(name) + except PluginError as e: + logger.warning(f"Skipping plugin {name} in listing: {e}") + continue if plugin_class: try: temp = plugin_class() diff --git a/shopdb/plugins/cli.py b/shopdb/plugins/cli.py index 1176db7..40083f8 100644 --- a/shopdb/plugins/cli.py +++ b/shopdb/plugins/cli.py @@ -577,6 +577,7 @@ def stamp_bundled(key_path, publisher, name): from .signing import (load_private_key, build_provenance, serialize_provenance, sign, PROVENANCE_NAME, PROVENANCE_SIG) + from .packaging import strip_bytecode pm = current_app.extensions.get('plugin_manager') if not pm: @@ -589,6 +590,9 @@ def stamp_bundled(key_path, publisher, name): for plugin_name in names: plugin_dir = plugins_dir / plugin_name + # Stamp a clean tree: no bytecode, so verify-at-load never trips on a + # cache the provenance does not cover. + strip_bytecode(plugin_dir) manifest = pm.loader.load_manifest(plugin_name) provenance = build_provenance( plugin_dir, manifest['name'], manifest['version'], publisher) @@ -743,7 +747,8 @@ def adopt_plugin(spec, pubkeys, force_downgrade): raise SystemExit(1) try: - resolved_version, artifact_name = resolve_version(index, name, version) + resolved_version, artifact_name, expected_sha = resolve_version( + index, name, version) except (KeyError, ValueError) as e: click.echo(click.style(f"{e}", fg='red')) raise SystemExit(1) @@ -763,7 +768,9 @@ def adopt_plugin(spec, pubkeys, force_downgrade): plugins_dir = Path(pm.loader.plugins_dir) click.echo(f"Verifying + unpacking {name} {resolved_version} ...") try: - unpack_verified(artifact_path, keys, plugins_dir, name) + unpack_verified(artifact_path, keys, plugins_dir, name, + expected_version=resolved_version, + expected_sha256=expected_sha) except ValueError as e: click.echo(click.style(f" verification failed: {e}", fg='red')) raise SystemExit(1) diff --git a/shopdb/plugins/loader.py b/shopdb/plugins/loader.py index 784d6d3..d732883 100644 --- a/shopdb/plugins/loader.py +++ b/shopdb/plugins/loader.py @@ -152,19 +152,38 @@ class PluginLoader: if name in self._plugin_classes: return self._plugin_classes[name] - plugin_module_path = self.plugins_dir / name / 'plugin.py' + plugin_dir = self.plugins_dir / name + plugin_module_path = plugin_dir / 'plugin.py' if not plugin_module_path.exists(): raise PluginNotFoundError( f'Plugin {name} plugin.py not found at {plugin_module_path}', 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 + enforcing = self.verifier is not None and self.verifier.require_signed + if enforcing: + strip_bytecode(plugin_dir) + if self.verifier is not None: + ok, reason = self.verifier.check(name) + if not ok: + raise PluginContractError( + f'Plugin {name} failed signature verification: {reason}', + plugin_name=name, + ) + 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) + with (no_bytecode() if enforcing else contextlib.nullcontext()): + 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}', @@ -200,16 +219,9 @@ class PluginLoader: manifest = self.load_manifest(name) self.check_contract_version(name, __contract_version__) - # verify-at-load: refuse to import plugin.py unless the tree matches - # a trusted signature (when the site enforces it). Runs BEFORE any - # plugin code is imported, so a tampered/unsigned plugin never runs. - if self.verifier is not None: - ok, reason = self.verifier.check(name) - if not ok: - raise PluginError( - f'Plugin {name} failed signature verification: {reason}', - plugin_name=name, - ) + # verify-at-load is enforced inside load_plugin_class (the single + # import choke point), so EVERY path that imports plugin code - + # load_plugin, discover_available, CLI introspection - fails closed. for dep in manifest.get('dependencies', []): if not self.registry.is_enabled(dep): diff --git a/shopdb/plugins/migrations.py b/shopdb/plugins/migrations.py index f71450b..9c8fbcc 100644 --- a/shopdb/plugins/migrations.py +++ b/shopdb/plugins/migrations.py @@ -21,6 +21,23 @@ class PluginMigrationManager: # Set by PluginManager.init_app; a PluginVerifier or None (no policy). self.verifier = None + def _verify_ok(self, plugin_name: str) -> bool: + """Strip bytecode + verify the tree before ANY alembic import. + + Alembic imports env.py and every versions/*.py module body (running + their top-level code) with full DB rights, so every method that reaches + alembic for a plugin must pass through here first (finding #3). + """ + if self.verifier is None or not self.verifier.require_signed: + return True + from .packaging import strip_bytecode + strip_bytecode(self.plugins_dir / plugin_name) + ok, reason = self.verifier.check(plugin_name) + if not ok: + logger.error( + "Refusing plugin migration code for %s: %s", plugin_name, reason) + return ok + def get_migrations_dir(self, plugin_name: str) -> Optional[Path]: """Get migrations directory for a plugin.""" migrations_dir = self.plugins_dir / plugin_name / 'migrations' @@ -38,15 +55,10 @@ class PluginMigrationManager: Uses flask db upgrade with the plugin's migrations directory. """ - # verify-at-migrate: never run a plugin's DDL (full DB rights) from an - # unverified tree when the site enforces signing. - if self.verifier is not None: - ok, reason = self.verifier.check(plugin_name) - if not ok: - logger.error( - "Refusing migrations for %s: signature verification failed " - "(%s)", plugin_name, reason) - return False + # verify-at-migrate: never run a plugin's migration code (full DB rights) + # from an unverified tree when the site enforces signing. + if not self._verify_ok(plugin_name): + return False migrations_dir = self.get_migrations_dir(plugin_name) @@ -58,6 +70,7 @@ class PluginMigrationManager: # Use alembic directly with plugin's migrations from alembic.config import Config from alembic import command + from .packaging import no_bytecode config = Config() config.set_main_option('script_location', str(migrations_dir)) @@ -69,7 +82,8 @@ class PluginMigrationManager: f'alembic_version_{plugin_name}' ) - command.upgrade(config, revision) + with no_bytecode(): + command.upgrade(config, revision) logger.info(f"Migrations completed for {plugin_name}") return True @@ -88,6 +102,9 @@ class PluginMigrationManager: revision: str = 'head' ) -> bool: """Run migrations via subprocess as fallback.""" + if not self._verify_ok(plugin_name): + return False + migrations_dir = self.get_migrations_dir(plugin_name) if not migrations_dir: return True @@ -95,7 +112,7 @@ class PluginMigrationManager: try: result = subprocess.run( [ - sys.executable, '-m', 'alembic', + sys.executable, '-B', '-m', 'alembic', '-c', str(migrations_dir / 'alembic.ini'), 'upgrade', revision ], @@ -103,7 +120,8 @@ class PluginMigrationManager: text=True, env={ **dict(__import__('os').environ), - 'DATABASE_URL': self.database_url + 'DATABASE_URL': self.database_url, + 'PYTHONDONTWRITEBYTECODE': '1', } ) @@ -125,6 +143,11 @@ class PluginMigrationManager: """ Downgrade/rollback plugin migrations. """ + # Downgrade runs the same env.py + version module code as upgrade, so it + # gets the same fail-closed verification (finding #3: this path had none). + if not self._verify_ok(plugin_name): + return False + migrations_dir = self.get_migrations_dir(plugin_name) if not migrations_dir: @@ -133,6 +156,7 @@ class PluginMigrationManager: try: from alembic.config import Config from alembic import command + from .packaging import no_bytecode config = Config() config.set_main_option('script_location', str(migrations_dir)) @@ -142,7 +166,8 @@ class PluginMigrationManager: f'alembic_version_{plugin_name}' ) - command.downgrade(config, revision) + with no_bytecode(): + command.downgrade(config, revision) logger.info(f"Downgrade completed for {plugin_name}") return True @@ -151,7 +176,14 @@ class PluginMigrationManager: return False def get_current_revision(self, plugin_name: str) -> Optional[str]: - """Get current migration revision for a plugin.""" + """Get current migration revision for a plugin. + + ScriptDirectory imports every versions/*.py to build the revision map, + so this reads (executes) plugin code and must verify first (finding #3). + """ + if not self._verify_ok(plugin_name): + return None + migrations_dir = self.get_migrations_dir(plugin_name) if not migrations_dir: return None @@ -159,12 +191,14 @@ class PluginMigrationManager: try: from alembic.config import Config from alembic.script import ScriptDirectory + from .packaging import no_bytecode config = Config() config.set_main_option('script_location', str(migrations_dir)) - script = ScriptDirectory.from_config(config) - return script.get_current_head() + with no_bytecode(): + script = ScriptDirectory.from_config(config) + return script.get_current_head() except Exception: return None diff --git a/shopdb/plugins/packaging.py b/shopdb/plugins/packaging.py index 2350d8e..0037dce 100644 --- a/shopdb/plugins/packaging.py +++ b/shopdb/plugins/packaging.py @@ -6,8 +6,11 @@ the exact provenance bytes). pack() builds and signs it; verify_artifact() and verify_dir() re-hash the contents and check the signature against trusted keys. """ +import contextlib import hashlib import json +import shutil +import sys import zipfile from pathlib import Path @@ -16,6 +19,39 @@ from . import signing ARTIFACT_SUFFIX = '.shopdbplugin' +def strip_bytecode(root) -> None: + """Remove every __pycache__ dir and .pyc/.pyo under root. + + A signed plugin tree must contain only the source that was hashed. Stripping + bytecode before verify + import guarantees the loader cannot execute a + planted or stale cache in place of the verified source. + """ + root = Path(root) + if not root.exists(): + return + for path in root.rglob('__pycache__'): + if path.is_dir(): + shutil.rmtree(path, ignore_errors=True) + for suffix in ('*.pyc', '*.pyo'): + for path in root.rglob(suffix): + try: + path.unlink() + except OSError: + pass + + +@contextlib.contextmanager +def no_bytecode(): + """Import/exec without writing .pyc, so a just-stripped tree stays clean and + a later verify does not trip over freshly written bytecode.""" + previous = sys.dont_write_bytecode + sys.dont_write_bytecode = True + try: + yield + finally: + sys.dont_write_bytecode = previous + + def pack(plugin_dir, private_key, publisher: str = '', out_dir=None, created: str = None) -> Path: """Build a signed artifact from a plugin directory. Returns its path. @@ -117,12 +153,12 @@ def verify_dir(plugin_dir, public_keys): for path in plugin_dir.rglob('*'): if path.is_file(): member_names.add(path.relative_to(plugin_dir).as_posix()) - # Drop the noise the packer also excludes, so an on-disk __pycache__ does - # not read as an "unexpected file". + # Ignore only non-executable dev noise. __pycache__/.pyc are NOT ignored - + # a planted bytecode cache must surface as an "unexpected file" so it cannot + # execute while escaping the hash map (finding #1). member_names = { m for m in member_names - if not any(part in signing._EXCLUDE_DIRS for part in Path(m).parts) - and Path(m).suffix not in signing._EXCLUDE_SUFFIXES + if not any(part in signing._VERIFY_EXCLUDE_DIRS for part in Path(m).parts) } def read_bytes(relpath): diff --git a/shopdb/plugins/shelf.py b/shopdb/plugins/shelf.py index cedace7..d3b96d4 100644 --- a/shopdb/plugins/shelf.py +++ b/shopdb/plugins/shelf.py @@ -8,6 +8,7 @@ ADOPT always re-verifies the artifact's own signed manifest, never trusting the index for anything security-bearing. """ +import hashlib import json import shutil import zipfile @@ -17,6 +18,14 @@ from packaging.version import Version, InvalidVersion from . import signing, packaging + +def _sha256_file(path) -> str: + digest = hashlib.sha256() + with open(path, 'rb') as handle: + for chunk in iter(lambda: handle.read(65536), b''): + digest.update(chunk) + return digest.hexdigest() + INDEX_NAME = 'shelf-index.json' INDEX_SIG = 'shelf-index.sig' STATE_NAME = 'shelf-state.json' @@ -41,6 +50,10 @@ def build_index(shelf_dir, private_key, serial: int, 'tier': manifest.get('tier', 'optional'), 'core_version': manifest.get('core_version', ''), 'artifact': artifact.name, + # Bind the entry to the artifact CONTENT, not just a filename, so an + # attacker who can rewrite the shelf folder cannot serve different + # (e.g. revoked) bytes under a live filename (finding #4). + 'sha256': _sha256_file(artifact), }) index = { @@ -111,10 +124,10 @@ def _safe_version(value): def resolve_version(index: dict, name: str, version: str = None): - """Pick the artifact filename for name[==version] from the index. + """Pick the artifact for name[==version] from the index. With no version, picks the highest non-revoked version. Returns - (version, artifact_filename) or raises KeyError/ValueError. + (version, artifact_filename, sha256_or_None) or raises KeyError/ValueError. """ candidates = index.get('plugins', {}).get(name) if not candidates: @@ -125,31 +138,46 @@ def resolve_version(index: dict, name: str, version: str = None): if entry['version'] == version: if is_revoked(index, name, version): raise ValueError(f'{name} {version} is revoked') - return version, entry['artifact'] + return version, entry['artifact'], entry.get('sha256') raise KeyError(f'{name} {version} not on the shelf') live = [e for e in candidates if not is_revoked(index, name, e['version'])] if not live: raise ValueError(f'every published version of {name} is revoked') best = max(live, key=lambda e: (_safe_version(e['version']) or Version('0'),)) - return best['version'], best['artifact'] + return best['version'], best['artifact'], best.get('sha256') -def unpack_verified(artifact_path, public_keys, plugins_dir, name: str): +def unpack_verified(artifact_path, public_keys, plugins_dir, name: str, + expected_version: str = None, expected_sha256: str = None): """Verify an artifact, then atomically place it at plugins_dir/name. - Verifies signature + every file hash in a staging area BEFORE it can be - imported; only a fully verified tree is moved into place. Returns the final - plugin directory. Raises ValueError with all problems on any failure. + Binds the delivered bytes to what the signed index resolved: the file digest + must match expected_sha256 (when given) and the artifact's OWN signed + manifest must carry name (and expected_version, when given). This stops a + revoked/other version being served under a live filename (finding #4). + Verifies signature + every file hash in staging BEFORE anything can import; + only a fully verified tree is moved into place. Raises ValueError on any + failure. """ plugins_dir = Path(plugins_dir) + if expected_sha256 is not None: + actual_sha = _sha256_file(artifact_path) + if actual_sha != expected_sha256: + raise ValueError( + 'artifact bytes do not match the signed index digest') + manifest, errors = packaging.verify_artifact(artifact_path, public_keys) if errors: raise ValueError('; '.join(errors)) if manifest.get('name') != name: raise ValueError( f"artifact manifest name '{manifest.get('name')}' != '{name}'") + if expected_version is not None and manifest.get('version') != expected_version: + raise ValueError( + f"artifact version '{manifest.get('version')}' != resolved " + f"'{expected_version}' (bytes served under wrong filename?)") staging = plugins_dir / STAGING_DIR staging.mkdir(parents=True, exist_ok=True) diff --git a/shopdb/plugins/signing.py b/shopdb/plugins/signing.py index be8627d..87989b8 100644 --- a/shopdb/plugins/signing.py +++ b/shopdb/plugins/signing.py @@ -23,9 +23,14 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey PROVENANCE_NAME = 'PROVENANCE.json' PROVENANCE_SIG = 'PROVENANCE.sig' -# Never packed, hashed, or counted as an unexpected artifact member. +# Never packed or hashed (a signed artifact carries source, never bytecode). _EXCLUDE_DIRS = {'__pycache__', '.pytest_cache', '.mypy_cache'} _EXCLUDE_SUFFIXES = {'.pyc', '.pyo'} +# Verification ignores only non-executable dev noise. It deliberately does NOT +# ignore __pycache__/.pyc: a planted bytecode cache would otherwise run while +# escaping the hash check (the set of verified bytes must not be smaller than +# the set of executed bytes). So verify flags any bytecode as an extra file. +_VERIFY_EXCLUDE_DIRS = {'.pytest_cache', '.mypy_cache', '.git'} _CHUNK = 65536 diff --git a/tests/test_plugin_packaging.py b/tests/test_plugin_packaging.py index 17ad103..9e6d23e 100644 --- a/tests/test_plugin_packaging.py +++ b/tests/test_plugin_packaging.py @@ -144,6 +144,10 @@ def test_verify_dir_round_trip(tmp_path, keypair): (plugin_dir / signing.PROVENANCE_SIG).write_bytes( signing.sign(priv, prov_bytes)) + # A real adopted/loaded tree carries no bytecode (packer excludes it, loader + # strips it); verify flags any that is present, so clear the test's planted + # __pycache__ before asserting a clean verify. + packaging.strip_bytecode(plugin_dir) _, errors = packaging.verify_dir(plugin_dir, [pub]) assert errors == [] diff --git a/tests/test_plugin_shelf.py b/tests/test_plugin_shelf.py index 122642e..6362077 100644 --- a/tests/test_plugin_shelf.py +++ b/tests/test_plugin_shelf.py @@ -99,9 +99,10 @@ def test_resolve_highest_non_revoked(tmp_path, keypair): shelf_dir = _build_shelf( tmp_path, priv, [('demo', '1.0.0'), ('demo', '1.2.0')]) index, _ = shelf.load_index(shelf_dir, [pub]) - version, artifact = shelf.resolve_version(index, 'demo') + version, artifact, sha = shelf.resolve_version(index, 'demo') assert version == '1.2.0' assert artifact == 'demo-1.2.0.shopdbplugin' + assert sha and len(sha) == 64 def test_resolve_skips_revoked(tmp_path, keypair): @@ -110,7 +111,7 @@ def test_resolve_skips_revoked(tmp_path, keypair): tmp_path, priv, [('demo', '1.0.0'), ('demo', '1.2.0')], revoked=[{'name': 'demo', 'version': '1.2.0'}]) index, _ = shelf.load_index(shelf_dir, [pub]) - version, _ = shelf.resolve_version(index, 'demo') + version, _, _ = shelf.resolve_version(index, 'demo') assert version == '1.0.0' @@ -178,3 +179,37 @@ def test_unpack_verified_wrong_name_refused(tmp_path, keypair): plugins_dir.mkdir() with pytest.raises(ValueError): shelf.unpack_verified(artifact, [pub], plugins_dir, 'somethingelse') + + +def test_revoked_bytes_under_live_filename_refused(tmp_path, keypair): + """Finding #4: an attacker swaps the genuinely-signed bytes of a revoked + 1.0.0 into the file named for the live 2.0.0. Adopt must refuse because the + artifact's own signed version (and content digest) do not match the resolved + version.""" + priv, pub = keypair + shelf_dir = tmp_path / 'shelf' + shelf_dir.mkdir() + # pack both versions + for version in ('1.0.0', '2.0.0'): + src = _write_plugin(tmp_path / 'src' / version, version=version) + packaging.pack(src, priv, out_dir=shelf_dir) + # index revokes 1.0.0, records true digests for 2.0.0 + shelf.build_index(shelf_dir, priv, 1, + revoked=[{'name': 'demo', 'version': '1.0.0'}]) + index, _ = shelf.load_index(shelf_dir, [pub]) + + resolved_version, artifact_name, expected_sha = shelf.resolve_version( + index, 'demo') + assert resolved_version == '2.0.0' + + # attacker overwrites the 2.0.0 file with the signed 1.0.0 bytes + live = shelf_dir / artifact_name + (shelf_dir / 'demo-1.0.0.shopdbplugin').replace(live) + + plugins_dir = tmp_path / 'live' + plugins_dir.mkdir() + with pytest.raises(ValueError): + shelf.unpack_verified(live, [pub], plugins_dir, 'demo', + expected_version=resolved_version, + expected_sha256=expected_sha) + assert not (plugins_dir / 'demo').exists() diff --git a/tests/test_plugin_verification.py b/tests/test_plugin_verification.py index 7daacbb..5966e84 100644 --- a/tests/test_plugin_verification.py +++ b/tests/test_plugin_verification.py @@ -158,6 +158,19 @@ def test_loader_loads_signed_when_enforced(tmp_path, app, db, keypair): assert plugin.meta.name == 'demo' +def test_load_plugin_class_refuses_unsigned_when_enforced(tmp_path): + """Finding #2: the gate is IN load_plugin_class, so the discover_available / + GET /api/plugins import path (which calls it directly, not via load_plugin) + also fails closed - an unsigned plugin.py is never imported.""" + plugins_dir = tmp_path / 'plugins' + _write_loadable_plugin(plugins_dir) + loader = PluginLoader(plugins_dir, PluginRegistry(tmp_path / 'plugins.json')) + loader.verifier = PluginVerifier(plugins_dir, require_signed=True, + trusted_key_paths=[]) + with pytest.raises(PluginError): + loader.load_plugin_class('demo') + + def test_migrate_refused_when_unverified(tmp_path): plugins_dir = tmp_path / 'plugins' _write_loadable_plugin(plugins_dir) @@ -165,3 +178,45 @@ def test_migrate_refused_when_unverified(tmp_path): manager.verifier = PluginVerifier(plugins_dir, require_signed=True, trusted_key_paths=[]) assert manager.run_plugin_migrations('demo') is False + + +def test_downgrade_refused_when_unverified(tmp_path): + """Finding #3: the downgrade path must fail closed like upgrade.""" + plugins_dir = tmp_path / 'plugins' + _write_loadable_plugin(plugins_dir) + manager = PluginMigrationManager(plugins_dir, 'sqlite://') + manager.verifier = PluginVerifier(plugins_dir, require_signed=True, + trusted_key_paths=[]) + assert manager.downgrade_plugin('demo') is False + + +def test_planted_bytecode_refused(tmp_path, app, db, keypair): + """Finding #1: a signed tree with an extra planted __pycache__/*.pyc must be + refused - the loader strips it before verify, and verify flags any bytecode + that survives, so only the signed source can run.""" + private_key, pub_path = keypair + plugins_dir = tmp_path / 'plugins' + pdir = _write_loadable_plugin(plugins_dir) + _stamp(pdir, private_key) + + # plant bytecode AFTER signing; provenance does not cover it + cache = pdir / '__pycache__' + cache.mkdir() + (cache / 'plugin.cpython-311.pyc').write_bytes(b'\x00malicious') + + verifier = PluginVerifier(plugins_dir, require_signed=True, + trusted_key_paths=[pub_path]) + + # verify BEFORE any strip: the planted bytecode is flagged as an extra file + ok, reason = verifier.check('demo') + assert not ok + assert 'unexpected file' in reason + + # the loader strips bytecode first, then verifies the clean source, so a + # legitimately signed plugin still loads + from shopdb.plugins.loader import PluginLoader + loader = PluginLoader(plugins_dir, PluginRegistry(tmp_path / 'plugins.json')) + loader.verifier = verifier + plugin = loader.load_plugin('demo', app, db) + assert plugin is not None + assert not (pdir / '__pycache__').exists()