shopdb-flask/tests/test_plugin_verification.py

"""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_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