"""Verify-at-load / verify-at-migrate enforcement tests (ADR-013 Phase 2). The security model rests on: with PLUGIN_REQUIRE_SIGNED set, a plugin only loads or migrates when its tree matches a trusted signature; unsigned, tampered, or wrong-key plugins are refused; and the dev-trust exemption only relaxes this under DEBUG/TESTING. """ import json import pytest from shopdb.exceptions import PluginError from shopdb.plugins import signing from shopdb.plugins.loader import PluginLoader from shopdb.plugins.migrations import PluginMigrationManager from shopdb.plugins.registry import PluginRegistry from shopdb.plugins.verification import PluginVerifier def _write_loadable_plugin(plugins_dir, name='demo'): """A minimal plugin whose plugin.py actually imports + instantiates.""" pdir = plugins_dir / name pdir.mkdir(parents=True, exist_ok=True) (pdir / 'manifest.json').write_text(json.dumps({ 'name': name, 'version': '1.0.0', 'description': 'demo', 'dependencies': [], 'core_version': '>=0.1.0,<1.0.0', 'api_prefix': f'/api/{name}', })) (pdir / '__init__.py').write_text('') (pdir / 'plugin.py').write_text( 'from shopdb.plugins.base import BasePlugin, PluginMeta\n\n' 'class ThePlugin(BasePlugin):\n' ' @property\n' ' def meta(self):\n' f' return PluginMeta(name={name!r}, version="1.0.0",\n' ' description="demo")\n' ' def get_blueprint(self):\n' ' return None\n' ' def get_models(self):\n' ' return []\n' ) return pdir def _stamp(pdir, private_key, name='demo', version='1.0.0'): provenance = signing.build_provenance(pdir, name, version, 'publisher') 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 / 'curator.pub' pub_path.write_bytes(public_pem) return signing.load_private_key(private_pem), str(pub_path) # --- PluginVerifier policy -------------------------------------------------- def test_enforcement_off_allows_unsigned(tmp_path): _write_loadable_plugin(tmp_path / 'plugins') verifier = PluginVerifier(tmp_path / 'plugins', require_signed=False) ok, _ = verifier.check('demo') assert ok def test_require_signed_without_keys_refuses_all(tmp_path): _write_loadable_plugin(tmp_path / 'plugins') verifier = PluginVerifier(tmp_path / 'plugins', require_signed=True, trusted_key_paths=[]) ok, reason = verifier.check('demo') assert not ok assert 'no trusted keys' in reason def test_signed_plugin_passes(tmp_path, keypair): private_key, pub_path = keypair pdir = _write_loadable_plugin(tmp_path / 'plugins') _stamp(pdir, private_key) verifier = PluginVerifier(tmp_path / 'plugins', require_signed=True, trusted_key_paths=[pub_path]) ok, reason = verifier.check('demo') assert ok, reason def test_tampered_plugin_refused(tmp_path, keypair): private_key, pub_path = keypair pdir = _write_loadable_plugin(tmp_path / 'plugins') _stamp(pdir, private_key) (pdir / 'plugin.py').write_text('# tampered after signing\n') verifier = PluginVerifier(tmp_path / 'plugins', require_signed=True, trusted_key_paths=[pub_path]) ok, reason = verifier.check('demo') assert not ok assert 'hash mismatch' in reason def test_wrong_key_refused(tmp_path, keypair): private_key, _ = keypair other_pub = tmp_path / 'attacker.pub' other_pub.write_bytes(signing.generate_keypair()[1]) pdir = _write_loadable_plugin(tmp_path / 'plugins') _stamp(pdir, private_key) verifier = PluginVerifier(tmp_path / 'plugins', require_signed=True, trusted_key_paths=[str(other_pub)]) ok, reason = verifier.check('demo') assert not ok assert 'signature does not match' in reason def test_dev_trust_dir_only_relaxes_in_dev(tmp_path, keypair): _, pub_path = keypair _write_loadable_plugin(tmp_path / 'plugins') # unsigned plugins_dir = tmp_path / 'plugins' dev = PluginVerifier(plugins_dir, require_signed=True, trusted_key_paths=[pub_path], dev_trust_dirs=[str(plugins_dir)], is_dev=True) ok, reason = dev.check('demo') assert ok and reason == 'dev-trusted' prod = PluginVerifier(plugins_dir, require_signed=True, trusted_key_paths=[pub_path], dev_trust_dirs=[str(plugins_dir)], is_dev=False) ok2, _ = prod.check('demo') assert not ok2 # --- verify-at-load and verify-at-migrate integration ----------------------- def test_loader_refuses_unsigned_when_enforced(tmp_path, app, db): plugins_dir = tmp_path / 'plugins' _write_loadable_plugin(plugins_dir) registry = PluginRegistry(tmp_path / 'plugins.json') loader = PluginLoader(plugins_dir, registry) loader.verifier = PluginVerifier(plugins_dir, require_signed=True, trusted_key_paths=[]) # TESTING app is strict -> the refusal surfaces as a raise. with pytest.raises(PluginError): loader.load_plugin('demo', app, db) def test_loader_loads_signed_when_enforced(tmp_path, app, db, keypair): private_key, pub_path = keypair plugins_dir = tmp_path / 'plugins' pdir = _write_loadable_plugin(plugins_dir) _stamp(pdir, private_key) registry = PluginRegistry(tmp_path / 'plugins.json') loader = PluginLoader(plugins_dir, registry) loader.verifier = PluginVerifier(plugins_dir, require_signed=True, trusted_key_paths=[pub_path]) plugin = loader.load_plugin('demo', app, db) assert plugin is not None 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) manager = PluginMigrationManager(plugins_dir, 'sqlite://') 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()