shopdb-flask/shopdb/plugins/packaging.py

"""Pack and verify signed plugin artifacts (.shopdbplugin) - ADR-013 Phase 1.

A .shopdbplugin is a zip of a plugin directory plus PROVENANCE.json (a per-file
SHA-256 map + metadata) and PROVENANCE.sig (a detached ed25519 signature over
the exact provenance bytes). pack() builds and signs it; verify_artifact() and
verify_dir() re-hash the contents and check the signature against trusted keys.
"""

import contextlib
import hashlib
import json
import shutil
import sys
import zipfile
from pathlib import Path

from . import signing

ARTIFACT_SUFFIX = '.shopdbplugin'


def strip_bytecode(root) -> None:
    """Remove every __pycache__ dir and .pyc/.pyo under root.

    A signed plugin tree must contain only the source that was hashed. Stripping
    bytecode before verify + import guarantees the loader cannot execute a
    planted or stale cache in place of the verified source.
    """
    root = Path(root)
    if not root.exists():
        return
    for path in root.rglob('__pycache__'):
        if path.is_dir():
            shutil.rmtree(path, ignore_errors=True)
    for suffix in ('*.pyc', '*.pyo'):
        for path in root.rglob(suffix):
            try:
                path.unlink()
            except OSError:
                pass


@contextlib.contextmanager
def no_bytecode():
    """Import/exec without writing .pyc, so a just-stripped tree stays clean and
    a later verify does not trip over freshly written bytecode."""
    previous = sys.dont_write_bytecode
    sys.dont_write_bytecode = True
    try:
        yield
    finally:
        sys.dont_write_bytecode = previous


def pack(plugin_dir, private_key, publisher: str = '',
         out_dir=None, created: str = None) -> Path:
    """Build a signed artifact from a plugin directory. Returns its path.

    Caller is expected to have validated the directory first (the CLI does).
    """
    plugin_dir = Path(plugin_dir)
    manifest = json.loads((plugin_dir / 'manifest.json').read_text())
    name = manifest['name']
    version = manifest['version']

    provenance = signing.build_provenance(
        plugin_dir, name, version, publisher, created)
    provenance_bytes = signing.serialize_provenance(provenance)
    signature = signing.sign(private_key, provenance_bytes)

    out_dir = Path(out_dir) if out_dir else plugin_dir.parent
    out_dir.mkdir(parents=True, exist_ok=True)
    artifact = out_dir / f'{name}-{version}{ARTIFACT_SUFFIX}'

    with zipfile.ZipFile(artifact, 'w', zipfile.ZIP_DEFLATED) as archive:
        for relpath in provenance['files']:
            archive.write(plugin_dir / relpath, relpath)
        archive.writestr(signing.PROVENANCE_NAME, provenance_bytes)
        archive.writestr(signing.PROVENANCE_SIG, signature)

    return artifact


def _verify_map(read_bytes, member_names, public_keys):
    """Shared core: given a way to read a member's bytes and the set of member
    names, verify signature + every file hash + no unexpected files.

    read_bytes(name) -> bytes. Returns (manifest_or_None, errors_list).
    """
    errors = []
    if signing.PROVENANCE_NAME not in member_names \
            or signing.PROVENANCE_SIG not in member_names:
        return None, ['missing PROVENANCE.json or PROVENANCE.sig']

    provenance_bytes = read_bytes(signing.PROVENANCE_NAME)
    signature = read_bytes(signing.PROVENANCE_SIG)

    if not public_keys:
        errors.append('no trusted keys supplied - signature unverifiable')
    elif not signing.verify(public_keys, provenance_bytes, signature):
        errors.append('signature does not match any trusted key')

    try:
        provenance = json.loads(provenance_bytes)
    except json.JSONDecodeError as exc:
        return None, errors + [f'PROVENANCE.json is not valid JSON: {exc}']

    filemap = provenance.get('files', {})
    for relpath, expected in filemap.items():
        if relpath not in member_names:
            errors.append(f'declared file missing: {relpath}')
            continue
        actual = hashlib.sha256(read_bytes(relpath)).hexdigest()
        if actual != expected:
            errors.append(f'hash mismatch: {relpath}')

    allowed = set(filemap) | {signing.PROVENANCE_NAME, signing.PROVENANCE_SIG}
    for member in member_names:
        if member not in allowed:
            errors.append(f'unexpected file not in provenance: {member}')

    manifest = None
    if 'manifest.json' in member_names:
        try:
            manifest = json.loads(read_bytes('manifest.json'))
        except json.JSONDecodeError:
            errors.append('manifest.json is not valid JSON')

    return manifest, errors


def verify_artifact(artifact_path, public_keys):
    """Verify a .shopdbplugin file. Returns (manifest_or_None, errors_list).

    errors empty = signature trusted AND every file hash intact AND no extra
    files. public_keys is a list of loaded ed25519 public keys (may be empty,
    which fails closed with an 'unverifiable' error).
    """
    with zipfile.ZipFile(artifact_path) as archive:
        member_names = {
            name for name in archive.namelist() if not name.endswith('/')}
        return _verify_map(archive.read, member_names, public_keys)


def verify_dir(plugin_dir, public_keys):
    """Verify an unpacked, adopted plugin directory that carries PROVENANCE
    files (used by verify-at-load/migrate in Phase 2). Same guarantees as
    verify_artifact.
    """
    plugin_dir = Path(plugin_dir)

    member_names = set()
    for path in plugin_dir.rglob('*'):
        if path.is_file():
            member_names.add(path.relative_to(plugin_dir).as_posix())
    # Ignore only non-executable dev noise. __pycache__/.pyc are NOT ignored -
    # a planted bytecode cache must surface as an "unexpected file" so it cannot
    # execute while escaping the hash map (finding #1).
    member_names = {
        m for m in member_names
        if not any(part in signing._VERIFY_EXCLUDE_DIRS for part in Path(m).parts)
    }

    def read_bytes(relpath):
        return (plugin_dir / relpath).read_bytes()

    return _verify_map(read_bytes, member_names, public_keys)