ADR-013 Phase 2: enforcement + signed shelf + adopt (adversarial-review hardened)

tests/test_plugin_verification.py

@@ -0,0 +1,222 @@
+"""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()