beea6c0c9f
Fourth review found the last import-path bypass: the is_dir() branch returned None for a name whose dir has no __init__.py, without checking a same-name sibling file. FileFinder loads a file over an init-less namespace dir, so an attacker could overwrite a signed foo.py with malicious bytes, mkdir an empty foo/ next to it (PROVENANCE untouched, still verifies), and any import of that name ran the unverified foo.py - RCE with only plugins/ write access. Fix: the dir-with-no-__init__.py branch no longer returns early; it falls through to the leaf .py hash gate and the non-source refuse check. Invariant: find_spec returns None for a plugins.* name ONLY where FileFinder would also find nothing on the same __path__. Everything else was confirmed sound this round: the owned plugins root, exec of exact verified bytes (never .pyc/.so), the extension/bytecode refusal, plugin.py read-once, the provenance signature gate, dev-exemption scoping, and #3/#4. Symlink, suffix-ordering, cache-lifecycle, and loader-internal angles cleared. 2 regression tests (tampered .py + sibling dir; unsigned .py + sibling dir). All 13 bundled plugins still load under enforcement; 1067 pass, naming green.
234 lines
8.9 KiB
Python
234 lines
8.9 KiB
Python
"""Import-guard tests (ADR-013 Phase 2 re-review fix).
|
|
|
|
The critical re-review finding: `plugins` is a normal package, so core code's
|
|
`from plugins.<name>.models import ...` bypassed the loader entirely and ran
|
|
unverified. These cover the sys.meta_path guard that verifies EVERY plugins.*
|
|
import from the exact bytes it execs.
|
|
"""
|
|
|
|
import importlib
|
|
import json
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from shopdb.plugins import signing
|
|
from shopdb.plugins.importguard import (PluginImportGuard,
|
|
PluginVerificationError,
|
|
install, uninstall, get_installed)
|
|
from shopdb.plugins.verification import PluginVerifier
|
|
|
|
|
|
def _write_pkg_plugin(plugins_dir, name, marker='source'):
|
|
"""A plugin with a models submodule that records which bytes executed."""
|
|
pdir = plugins_dir / name
|
|
(pdir / 'models').mkdir(parents=True)
|
|
(pdir / 'manifest.json').write_text(json.dumps({
|
|
'name': name, 'version': '1.0.0', 'description': 'd',
|
|
'core_version': '>=0.1.0,<1.0.0', 'api_prefix': f'/api/{name}',
|
|
}))
|
|
(pdir / '__init__.py').write_text('')
|
|
(pdir / 'plugin.py').write_text('# plugin\n')
|
|
(pdir / 'models' / '__init__.py').write_text(f'MARKER = {marker!r}\n')
|
|
return pdir
|
|
|
|
|
|
def _stamp(pdir, private_key, name):
|
|
provenance = signing.build_provenance(pdir, name, '1.0.0', 'pub')
|
|
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 / 'k.pub'
|
|
pub_path.write_bytes(public_pem)
|
|
return signing.load_private_key(private_pem), str(pub_path)
|
|
|
|
|
|
def _verifier(plugins_dir, pub_path):
|
|
return PluginVerifier(plugins_dir, require_signed=True,
|
|
trusted_key_paths=[pub_path])
|
|
|
|
|
|
@pytest.fixture
|
|
def guarded(tmp_path, keypair):
|
|
"""Install a guard, yield (guard, plugins_dir, private_key, name), clean up
|
|
sys.modules + sys.meta_path afterward so tests do not leak."""
|
|
private_key, pub_path = keypair
|
|
plugins_dir = tmp_path / 'plugins'
|
|
plugins_dir.mkdir()
|
|
name = 'guardtestpkg'
|
|
guard = install(plugins_dir, _verifier(plugins_dir, pub_path))
|
|
try:
|
|
yield guard, plugins_dir, private_key, name
|
|
finally:
|
|
uninstall()
|
|
for mod in list(sys.modules):
|
|
if mod == f'plugins.{name}' or mod.startswith(f'plugins.{name}.'):
|
|
del sys.modules[mod]
|
|
|
|
|
|
# --- direct find_spec / verified_source -------------------------------------
|
|
|
|
def test_verified_source_signed_ok_tampered_raises(guarded):
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
|
|
assert guard.verified_source(name, 'plugin.py') == b'# plugin\n'
|
|
|
|
(pdir / 'plugin.py').write_text('# tampered\n')
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.verified_source(name, 'plugin.py')
|
|
|
|
|
|
def test_find_spec_refuses_unsigned_plugin(guarded):
|
|
guard, plugins_dir, priv, name = guarded
|
|
_write_pkg_plugin(plugins_dir, name) # no provenance stamped
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.find_spec(f'plugins.{name}.models')
|
|
|
|
|
|
def test_find_spec_refuses_tampered_submodule(guarded):
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
# tamper the submodule after signing
|
|
(pdir / 'models' / '__init__.py').write_text('MARKER = "evil"\n')
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.find_spec(f'plugins.{name}.models')
|
|
|
|
|
|
# --- real import through the installed guard --------------------------------
|
|
|
|
def test_core_style_submodule_import_is_verified(guarded):
|
|
"""`from plugins.<name>.models import MARKER` runs the SIGNED source."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name, marker='source')
|
|
_stamp(pdir, priv, name)
|
|
|
|
module = importlib.import_module(f'plugins.{name}.models')
|
|
assert module.MARKER == 'source'
|
|
|
|
|
|
def test_planted_pyc_is_never_consulted(guarded):
|
|
"""A planted __pycache__/*.pyc must not run: the guard loads verified source
|
|
only, so a garbage bytecode cache is simply ignored (finding #1/#2)."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name, marker='source')
|
|
_stamp(pdir, priv, name)
|
|
|
|
cache = pdir / 'models' / '__pycache__'
|
|
cache.mkdir()
|
|
# garbage bytecode: if the loader ever honored it, import would crash
|
|
for pyc in (f'__init__.cpython-{sys.version_info.major}'
|
|
f'{sys.version_info.minor}.pyc',):
|
|
(cache / pyc).write_bytes(b'\x00\x00\x00\x00garbage-not-valid-bytecode')
|
|
|
|
module = importlib.import_module(f'plugins.{name}.models')
|
|
assert module.MARKER == 'source' # source ran, pyc ignored
|
|
|
|
|
|
def test_tampered_submodule_import_raises(guarded):
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
(pdir / 'models' / '__init__.py').write_text('MARKER = "evil"\n')
|
|
with pytest.raises(ImportError):
|
|
importlib.import_module(f'plugins.{name}.models')
|
|
|
|
|
|
def test_planted_extension_module_refused(guarded):
|
|
"""Finding (round 3): a planted .so standing in for a would-be submodule must
|
|
be refused, not handed to the stdlib ExtensionFileLoader unverified."""
|
|
import importlib.machinery
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
|
|
ext = importlib.machinery.EXTENSION_SUFFIXES[0]
|
|
(pdir / 'models' / f'evil{ext}').write_bytes(b'\x7fELF-not-really')
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.find_spec(f'plugins.{name}.models.evil')
|
|
|
|
|
|
def test_planted_sourceless_pyc_refused(guarded):
|
|
"""A sourceless .pyc for a name with no .py must be refused."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
|
|
(pdir / 'models' / 'evil.pyc').write_bytes(b'\x00garbage')
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.find_spec(f'plugins.{name}.models.evil')
|
|
|
|
|
|
def test_tampered_py_with_sibling_dir_refused(guarded):
|
|
"""Round 4: a signed foo.py tampered + an empty foo/ dir next to it must be
|
|
refused. FileFinder would load the file over the init-less dir, so the guard
|
|
must hash-gate the sibling .py instead of returning None for the dir."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
(pdir / 'models' / 'device.py').write_text('MARKER = "signed"\n')
|
|
_stamp(pdir, priv, name)
|
|
|
|
# attacker tampers the signed device.py and adds an empty sibling device/
|
|
(pdir / 'models' / 'device.py').write_text('MARKER = "evil"\n')
|
|
(pdir / 'models' / 'device').mkdir()
|
|
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.find_spec(f'plugins.{name}.models.device')
|
|
|
|
|
|
def test_unsigned_py_with_sibling_dir_refused(guarded):
|
|
"""Symmetric variant: a never-signed evil.py next to evil/ is refused too."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
|
|
(pdir / 'models' / 'evil.py').write_text('MARKER = "evil"\n')
|
|
(pdir / 'models' / 'evil').mkdir()
|
|
|
|
with pytest.raises(PluginVerificationError):
|
|
guard.find_spec(f'plugins.{name}.models.evil')
|
|
|
|
|
|
def test_absent_module_defers(guarded):
|
|
"""A genuinely-absent module returns None (normal ModuleNotFoundError)."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
pdir = _write_pkg_plugin(plugins_dir, name)
|
|
_stamp(pdir, priv, name)
|
|
assert guard.find_spec(f'plugins.{name}.models.notthere') is None
|
|
|
|
|
|
def test_top_level_plugins_body_neutralized(guarded):
|
|
"""Finding (round 3): the top-level plugins/__init__.py runs first and is in
|
|
no provenance; the guard must exec an EMPTY body so an attacker-overwritten
|
|
plugins/__init__.py cannot execute."""
|
|
guard, plugins_dir, priv, name = guarded
|
|
(plugins_dir / '__init__.py').write_text('raise RuntimeError("attacker code")')
|
|
spec = guard.find_spec('plugins')
|
|
assert spec is not None
|
|
# the loader carries empty source, not the on-disk body
|
|
assert spec.loader._source == b''
|
|
assert str(plugins_dir) in spec.submodule_search_locations
|
|
|
|
|
|
def test_install_uninstall_idempotent(tmp_path, keypair):
|
|
_, pub_path = keypair
|
|
plugins_dir = tmp_path / 'plugins'
|
|
plugins_dir.mkdir()
|
|
verifier = _verifier(plugins_dir, pub_path)
|
|
try:
|
|
install(plugins_dir, verifier)
|
|
install(plugins_dir, verifier) # replaces, does not stack
|
|
assert sum(isinstance(f, PluginImportGuard) for f in sys.meta_path) == 1
|
|
assert get_installed() is not None
|
|
finally:
|
|
uninstall()
|
|
assert get_installed() is None
|