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.
158 lines
6.0 KiB
Python
158 lines
6.0 KiB
Python
"""Signed plugin artifact packaging tests (ADR-013 Phase 1).
|
|
|
|
Covers the provenance + ed25519 signing that the marketplace trust model rests
|
|
on: a valid signature over an intact file set verifies, and every tampering
|
|
path (flipped file byte, wrong key, missing/extra file, no key) fails closed.
|
|
"""
|
|
|
|
import json
|
|
import zipfile
|
|
|
|
import pytest
|
|
|
|
from shopdb.plugins import signing, packaging
|
|
|
|
|
|
def _write_plugin(plugin_dir, name='demo', version='1.0.0', deps=None):
|
|
plugin_dir.mkdir(parents=True, exist_ok=True)
|
|
(plugin_dir / 'manifest.json').write_text(json.dumps({
|
|
'name': name, 'version': version, 'description': 'demo',
|
|
'dependencies': deps or [], 'core_version': '>=0.1.0,<1.0.0',
|
|
'api_prefix': f'/api/{name}',
|
|
}))
|
|
(plugin_dir / 'plugin.py').write_text('# demo plugin\n')
|
|
(plugin_dir / 'models').mkdir()
|
|
(plugin_dir / 'models' / '__init__.py').write_text('')
|
|
# noise that must NOT be packed or hashed
|
|
(plugin_dir / '__pycache__').mkdir()
|
|
(plugin_dir / '__pycache__' / 'x.pyc').write_bytes(b'\x00\x01')
|
|
return plugin_dir
|
|
|
|
|
|
# --- signing primitives -----------------------------------------------------
|
|
|
|
def test_sign_verify_round_trip():
|
|
private_pem, public_pem = signing.generate_keypair()
|
|
priv = signing.load_private_key(private_pem)
|
|
pub = signing.load_public_key(public_pem)
|
|
|
|
data = b'provenance bytes'
|
|
sig = signing.sign(priv, data)
|
|
|
|
assert signing.verify([pub], data, sig) is True
|
|
assert signing.verify([pub], b'other bytes', sig) is False
|
|
|
|
|
|
def test_wrong_key_does_not_verify():
|
|
priv = signing.load_private_key(signing.generate_keypair()[0])
|
|
other_pub = signing.load_public_key(signing.generate_keypair()[1])
|
|
sig = signing.sign(priv, b'data')
|
|
assert signing.verify([other_pub], b'data', sig) is False
|
|
|
|
|
|
def test_provenance_excludes_noise(tmp_path):
|
|
plugin_dir = _write_plugin(tmp_path / 'demo')
|
|
filemap = signing.build_file_map(plugin_dir)
|
|
assert 'manifest.json' in filemap
|
|
assert 'models/__init__.py' in filemap
|
|
assert not any('__pycache__' in k or k.endswith('.pyc') for k in filemap)
|
|
assert signing.PROVENANCE_NAME not in filemap
|
|
|
|
|
|
def test_serialize_provenance_is_order_independent():
|
|
a = {'name': 'x', 'version': '1', 'files': {'b': '2', 'a': '1'}}
|
|
b = {'files': {'a': '1', 'b': '2'}, 'version': '1', 'name': 'x'}
|
|
assert signing.serialize_provenance(a) == signing.serialize_provenance(b)
|
|
|
|
|
|
# --- pack + verify ----------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def keypair():
|
|
private_pem, public_pem = signing.generate_keypair()
|
|
return (signing.load_private_key(private_pem),
|
|
signing.load_public_key(public_pem))
|
|
|
|
|
|
def test_pack_then_verify_ok(tmp_path, keypair):
|
|
priv, pub = keypair
|
|
plugin_dir = _write_plugin(tmp_path / 'demo')
|
|
|
|
artifact = packaging.pack(plugin_dir, priv, publisher='site-a',
|
|
out_dir=tmp_path / 'dist')
|
|
assert artifact.name == 'demo-1.0.0.shopdbplugin'
|
|
|
|
manifest, errors = packaging.verify_artifact(artifact, [pub])
|
|
assert errors == []
|
|
assert manifest['name'] == 'demo'
|
|
|
|
|
|
def test_verify_fails_with_no_keys(tmp_path, keypair):
|
|
priv, _ = keypair
|
|
artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv,
|
|
out_dir=tmp_path / 'dist')
|
|
_, errors = packaging.verify_artifact(artifact, [])
|
|
assert any('unverifiable' in e for e in errors)
|
|
|
|
|
|
def test_verify_fails_with_wrong_key(tmp_path, keypair):
|
|
priv, _ = keypair
|
|
wrong_pub = signing.load_public_key(signing.generate_keypair()[1])
|
|
artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv,
|
|
out_dir=tmp_path / 'dist')
|
|
_, errors = packaging.verify_artifact(artifact, [wrong_pub])
|
|
assert any('signature does not match' in e for e in errors)
|
|
|
|
|
|
def test_verify_detects_tampered_file(tmp_path, keypair):
|
|
priv, pub = keypair
|
|
artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv,
|
|
out_dir=tmp_path / 'dist')
|
|
|
|
# Rewrite the zip with one member's bytes changed, keeping provenance + sig.
|
|
with zipfile.ZipFile(artifact) as zin:
|
|
members = {n: zin.read(n) for n in zin.namelist()}
|
|
members['manifest.json'] += b'tampered'
|
|
with zipfile.ZipFile(artifact, 'w', zipfile.ZIP_DEFLATED) as zout:
|
|
for name, data in members.items():
|
|
zout.writestr(name, data)
|
|
|
|
_, errors = packaging.verify_artifact(artifact, [pub])
|
|
assert any('hash mismatch' in e for e in errors)
|
|
|
|
|
|
def test_verify_detects_extra_file(tmp_path, keypair):
|
|
priv, pub = keypair
|
|
artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv,
|
|
out_dir=tmp_path / 'dist')
|
|
with zipfile.ZipFile(artifact, 'a', zipfile.ZIP_DEFLATED) as z:
|
|
z.writestr('sneaky.py', 'print("surprise")')
|
|
|
|
_, errors = packaging.verify_artifact(artifact, [pub])
|
|
assert any('unexpected file' in e for e in errors)
|
|
|
|
|
|
def test_verify_dir_round_trip(tmp_path, keypair):
|
|
"""An unpacked directory carrying PROVENANCE files verifies (Phase 2 uses
|
|
this for verify-at-load)."""
|
|
priv, pub = keypair
|
|
plugin_dir = _write_plugin(tmp_path / 'demo')
|
|
|
|
provenance = signing.build_provenance(plugin_dir, 'demo', '1.0.0', 'site-a')
|
|
prov_bytes = signing.serialize_provenance(provenance)
|
|
(plugin_dir / signing.PROVENANCE_NAME).write_bytes(prov_bytes)
|
|
(plugin_dir / signing.PROVENANCE_SIG).write_bytes(
|
|
signing.sign(priv, prov_bytes))
|
|
|
|
# A real adopted/loaded tree carries no bytecode (packer excludes it, loader
|
|
# strips it); verify flags any that is present, so clear the test's planted
|
|
# __pycache__ before asserting a clean verify.
|
|
packaging.strip_bytecode(plugin_dir)
|
|
_, errors = packaging.verify_dir(plugin_dir, [pub])
|
|
assert errors == []
|
|
|
|
# Tamper a file on disk -> hash mismatch.
|
|
(plugin_dir / 'plugin.py').write_text('# changed\n')
|
|
_, errors = packaging.verify_dir(plugin_dir, [pub])
|
|
assert any('hash mismatch' in e for e in errors)
|