shopdb-flask/shopdb/plugins/importguard.py

"""Import-time verification for ALL plugin code (ADR-013 Phase 2 hardening).

`plugins` is a normal importable package: core request handlers do
`from plugins.<name>.models import ...` through the standard import system,
which never passes through the plugin loader. Gating only load_plugin_class
(plugin.py) therefore left every submodule import unverified - a planted
plugins/<name>/models/*.py executed on an ordinary HTTP request, and a planted
__pycache__/*.pyc ran from a cached read. There is no "single choke point" in
the loader; the choke point is the import system itself.

This installs a sys.meta_path finder that intercepts every `plugins.<name>.*`
import, verifies the plugin's signed provenance once, then verifies each module
file against that provenance and EXECUTES THE EXACT BYTES IT HASHED (read once,
compile, exec) - never a .pyc, never a re-opened file. That closes the submodule
bypass (#1/#2) and the verify-vs-exec TOCTOU (#3) for the import path together.

Installed only under enforcement (PLUGIN_REQUIRE_SIGNED). When off, the finder
is absent and imports behave exactly as before.
"""

import hashlib
import importlib.abc
import importlib.util
import json
import sys
from pathlib import Path

from . import signing


class PluginVerificationError(ImportError):
    """Raised when a plugins.* module is not covered by a trusted signature."""


class _VerifiedSourceLoader(importlib.abc.Loader):
    """Execs source bytes that were already hash-verified (no re-open, no .pyc)."""

    def __init__(self, filepath, source_bytes, is_package, search_locations):
        self._filepath = str(filepath)
        self._source = source_bytes
        self._is_package = is_package
        self._search = search_locations

    def create_module(self, spec):
        return None  # default module creation

    def exec_module(self, module):
        code = compile(self._source, self._filepath, 'exec')
        exec(code, module.__dict__)

    def is_package(self, fullname):
        return self._is_package


class PluginImportGuard(importlib.abc.MetaPathFinder):
    """Verifies and loads every plugins.<name>.* module from signed bytes."""

    def __init__(self, plugins_dir, verifier):
        self.plugins_dir = Path(plugins_dir)
        self.verifier = verifier
        self._filemaps = {}  # plugin name -> verified {relpath: sha256}

    def _filemap(self, name):
        """Verify the plugin's provenance signature ONCE, cache its file map."""
        if name in self._filemaps:
            return self._filemaps[name]
        plugin_dir = self.plugins_dir / name
        provenance_path = plugin_dir / signing.PROVENANCE_NAME
        signature_path = plugin_dir / signing.PROVENANCE_SIG
        if not provenance_path.exists() or not signature_path.exists():
            raise PluginVerificationError(
                f'plugin {name} has no provenance; refusing import under '
                f'enforcement')
        provenance_bytes = provenance_path.read_bytes()
        signature = signature_path.read_bytes()
        if not self.verifier._keys or not signing.verify(
                self.verifier._keys, provenance_bytes, signature):
            raise PluginVerificationError(
                f'plugin {name} provenance signature is not trusted')
        filemap = json.loads(provenance_bytes).get('files', {})
        self._filemaps[name] = filemap
        return filemap

    def _module_file(self, name, fullname):
        """(filepath, is_package, search_locations) for a plugins.<name>.* module,
        or (None, ...) when it is not a source module we should load."""
        plugin_dir = self.plugins_dir / name
        tail = fullname.split('.')[2:]  # components after plugins.<name>
        base = plugin_dir.joinpath(*tail) if tail else plugin_dir
        if base.is_dir():
            return base / '__init__.py', True, [str(base)]
        source = base.with_suffix('.py')
        if source.exists():
            return source, False, None
        return None, False, None

    def find_spec(self, fullname, path=None, target=None):
        # Only our package; the empty top-level `plugins` package loads normally.
        if fullname != 'plugins' and not fullname.startswith('plugins.'):
            return None
        if fullname == 'plugins':
            return None
        name = fullname.split('.')[1]
        plugin_dir = self.plugins_dir / name

        # Dev/external-repo plugins are exempt (only ever under DEBUG/TESTING).
        if self.verifier._dev_exempt(plugin_dir):
            return None

        filemap = self._filemap(name)  # verify signature (raises on failure)
        filepath, is_package, search = self._module_file(name, fullname)
        if filepath is None or not filepath.exists():
            # A missing __init__.py (namespace pkg) or non-python target: let the
            # normal machinery decide. Any .py it would run is covered above.
            return None

        relpath = filepath.relative_to(plugin_dir).as_posix()
        source = filepath.read_bytes()
        expected = filemap.get(relpath)
        if expected is None or hashlib.sha256(source).hexdigest() != expected:
            raise PluginVerificationError(
                f'plugin {name} module {relpath} is not covered by a trusted '
                f'signature')

        loader = _VerifiedSourceLoader(filepath, source, is_package, search)
        spec = importlib.util.spec_from_loader(
            fullname, loader, is_package=is_package)
        if is_package:
            spec.submodule_search_locations = search
        return spec

    def verified_source(self, name, relpath):
        """Return hash-verified bytes of one plugin file (read once), for callers
        that load a file explicitly (load_plugin_class + plugin.py). Raises on
        any mismatch. Closes the TOCTOU on that file: the caller execs exactly
        these bytes."""
        filemap = self._filemap(name)
        source = (self.plugins_dir / name / relpath).read_bytes()
        expected = filemap.get(relpath)
        if expected is None or hashlib.sha256(source).hexdigest() != expected:
            raise PluginVerificationError(
                f'plugin {name} file {relpath} is not covered by a trusted '
                f'signature')
        return source


def get_installed():
    """Return the installed guard, or None."""
    for finder in sys.meta_path:
        if isinstance(finder, PluginImportGuard):
            return finder
    return None


def install(plugins_dir, verifier):
    """Install the guard at the FRONT of sys.meta_path (idempotent, replaces any
    prior guard so a re-init picks up new config)."""
    uninstall()
    guard = PluginImportGuard(plugins_dir, verifier)
    sys.meta_path.insert(0, guard)
    return guard


def uninstall():
    """Remove any installed guard (used on teardown / when enforcement is off)."""
    sys.meta_path[:] = [
        f for f in sys.meta_path if not isinstance(f, PluginImportGuard)]