4c1d96b6ba
Exports caller_is_role on the plugin authorization surface. Additive name, so a minor bump. Scoped personal tokens are always refused by it, matching require_role, which stops a plugin from accidentally letting a scoped token inherit role authority from its owner. Registers the tables added by the endpoint lock and policy hold work (endpointlocks, policyactionholds) plus the endpoint health tables in the alembic template, so plugin migrations see them and the ownership guard does not report them as unclaimed.
237 lines
9.1 KiB
Python
237 lines
9.1 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('')
|
|
# write_bytes, not write_text: on Windows write_text translates \n to
|
|
# \r\n, which changes the bytes that get signed and hashed and makes the
|
|
# verified_source assertion below fail for a purely platform reason.
|
|
(pdir / 'plugin.py').write_bytes(b'# 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_bytes(b'# 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
|