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.
142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""Ed25519 signing + provenance for plugin artifacts (ADR-013 Phase 1).
|
|
|
|
A plugin's provenance is a per-file SHA-256 map plus metadata (name, version,
|
|
publisher, created). The detached signature covers the EXACT serialized
|
|
provenance bytes, so verifying a plugin is: re-hash its files, re-serialize the
|
|
provenance the same way, and check the signature. Any changed byte in any file
|
|
changes a hash, which changes the serialized provenance, which fails the
|
|
signature. The signature proves the artifact is exactly what a curator reviewed
|
|
and signed - nothing about what the code does.
|
|
|
|
Uses the cryptography package (already a dependency for MySQL 8 auth).
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from cryptography.exceptions import InvalidSignature
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
PROVENANCE_NAME = 'PROVENANCE.json'
|
|
PROVENANCE_SIG = 'PROVENANCE.sig'
|
|
|
|
# Never packed or hashed (a signed artifact carries source, never bytecode).
|
|
_EXCLUDE_DIRS = {'__pycache__', '.pytest_cache', '.mypy_cache'}
|
|
_EXCLUDE_SUFFIXES = {'.pyc', '.pyo'}
|
|
# Verification ignores only non-executable dev noise. It deliberately does NOT
|
|
# ignore __pycache__/.pyc: a planted bytecode cache would otherwise run while
|
|
# escaping the hash check (the set of verified bytes must not be smaller than
|
|
# the set of executed bytes). So verify flags any bytecode as an extra file.
|
|
_VERIFY_EXCLUDE_DIRS = {'.pytest_cache', '.mypy_cache', '.git'}
|
|
_CHUNK = 65536
|
|
|
|
|
|
def generate_keypair():
|
|
"""Return (private_pem, public_pem) as PEM bytes for a new ed25519 key."""
|
|
private_key = Ed25519PrivateKey.generate()
|
|
private_pem = private_key.private_bytes(
|
|
serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.PKCS8,
|
|
serialization.NoEncryption(),
|
|
)
|
|
public_pem = private_key.public_key().public_bytes(
|
|
serialization.Encoding.PEM,
|
|
serialization.PublicFormat.SubjectPublicKeyInfo,
|
|
)
|
|
return private_pem, public_pem
|
|
|
|
|
|
def load_private_key(pem_bytes: bytes):
|
|
return serialization.load_pem_private_key(pem_bytes, password=None)
|
|
|
|
|
|
def load_public_key(pem_bytes: bytes):
|
|
return serialization.load_pem_public_key(pem_bytes)
|
|
|
|
|
|
def load_trusted_keys(pem_paths):
|
|
"""Load public keys from a list of PEM file paths. Missing/unreadable paths
|
|
are skipped (they simply cannot vouch for a signature)."""
|
|
keys = []
|
|
for path in pem_paths or []:
|
|
try:
|
|
keys.append(load_public_key(Path(path).read_bytes()))
|
|
except (OSError, ValueError):
|
|
continue
|
|
return keys
|
|
|
|
|
|
def _packable(plugin_dir: Path):
|
|
"""Yield the plugin's real files (sorted, posix relpaths excluded from noise)."""
|
|
for path in sorted(plugin_dir.rglob('*')):
|
|
if not path.is_file():
|
|
continue
|
|
rel = path.relative_to(plugin_dir)
|
|
if any(part in _EXCLUDE_DIRS for part in rel.parts):
|
|
continue
|
|
if path.suffix in _EXCLUDE_SUFFIXES:
|
|
continue
|
|
if path.name in (PROVENANCE_NAME, PROVENANCE_SIG):
|
|
continue
|
|
yield rel.as_posix(), path
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with open(path, 'rb') as handle:
|
|
for chunk in iter(lambda: handle.read(_CHUNK), b''):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def build_file_map(plugin_dir) -> dict:
|
|
"""{posix relpath: sha256hex} for every packable file, sorted."""
|
|
plugin_dir = Path(plugin_dir)
|
|
return {relpath: _sha256_file(path) for relpath, path in _packable(plugin_dir)}
|
|
|
|
|
|
def build_provenance(plugin_dir, name: str, version: str,
|
|
publisher: str = '', created: str = None) -> dict:
|
|
"""Provenance dict. `created` defaults to now (UTC, second precision)."""
|
|
if created is None:
|
|
created = datetime.now(timezone.utc).replace(
|
|
microsecond=0, tzinfo=None).isoformat()
|
|
return {
|
|
'name': name,
|
|
'version': version,
|
|
'publisher': publisher or '',
|
|
'created': created,
|
|
'files': build_file_map(plugin_dir),
|
|
}
|
|
|
|
|
|
def serialize_provenance(provenance: dict) -> bytes:
|
|
"""Canonical bytes that get signed AND stored, so sign and verify agree.
|
|
|
|
sort_keys + compact separators make this deterministic; the same dict always
|
|
serializes to the same bytes regardless of insertion order.
|
|
"""
|
|
return json.dumps(
|
|
provenance, sort_keys=True, separators=(',', ':')).encode('utf-8')
|
|
|
|
|
|
def sign(private_key, data: bytes) -> bytes:
|
|
return private_key.sign(data)
|
|
|
|
|
|
def verify(public_keys, data: bytes, signature: bytes) -> bool:
|
|
"""True if the signature validates against ANY trusted public key.
|
|
|
|
Multiple keys support overlap during key rotation.
|
|
"""
|
|
for key in public_keys:
|
|
try:
|
|
key.verify(signature, data)
|
|
return True
|
|
except InvalidSignature:
|
|
continue
|
|
return False
|