55a6f1b8d3
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.
223 lines
8.8 KiB
Python
223 lines
8.8 KiB
Python
"""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()
|