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.machinery
import importlib.util
import json
import sys
from pathlib import Path

from . import signing

# Non-source files the stdlib import system would execute for a module name.
# The guard loads ONLY verified .py source, so any of these under a plugin is a
# module the guard cannot vouch for and must refuse (a planted .so or a
# sourceless .pyc standing in for a deleted, signed .py).
_UNVERIFIABLE_SUFFIXES = tuple(
    importlib.machinery.EXTENSION_SUFFIXES
    + importlib.machinery.BYTECODE_SUFFIXES)


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

    def get_filename(self, fullname):
        # gives the loaded module a __file__ so Flask can resolve blueprint
        # root paths (spec.origin is set from this).
        return self._filepath

    def get_source(self, fullname):
        return self._source.decode('utf-8', 'replace')


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 _refuse_if_unverifiable_present(self, directory, leaf, name):
        """Fail closed if a non-.py importable file (.so / sourceless .pyc) for
        `leaf` sits in `directory`. Reserving None for genuinely-absent names is
        what stops the stdlib loading a planted extension/bytecode unverified."""
        for suffix in _UNVERIFIABLE_SUFFIXES:
            if (directory / f'{leaf}{suffix}').exists():
                raise PluginVerificationError(
                    f'plugin {name} module {leaf}{suffix} is not loadable from '
                    f'trusted source; refusing')

    def _source_spec(self, name, fullname, filepath, is_package, search):
        plugin_dir = self.plugins_dir / name
        relpath = filepath.relative_to(plugin_dir).as_posix()
        source = filepath.read_bytes()
        expected = self._filemap(name).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_from_file_location sets origin -> the module gets __file__/__path__,
        # which Flask needs for blueprint root paths; the loader still execs the
        # verified in-memory bytes, never re-reading the file.
        return importlib.util.spec_from_file_location(
            fullname, str(filepath), loader=loader,
            submodule_search_locations=search if is_package else None)

    def find_spec(self, fullname, path=None, target=None):
        if fullname != 'plugins' and not fullname.startswith('plugins.'):
            return None

        # The top-level `plugins` package is in no plugin's provenance, yet
        # Python imports it before any guarded submodule. Own it: execute an
        # EMPTY package body so an attacker-overwritten plugins/__init__.py never
        # runs, and point package search at the plugins dir.
        if fullname == 'plugins':
            init_path = self.plugins_dir / '__init__.py'
            loader = _VerifiedSourceLoader(
                init_path, b'', True, [str(self.plugins_dir)])
            return importlib.util.spec_from_file_location(
                'plugins', str(init_path), loader=loader,
                submodule_search_locations=[str(self.plugins_dir)])

        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

        self._filemap(name)  # verify the plugin signature (raises on failure)

        tail = fullname.split('.')[2:]  # components after plugins.<name>
        base = plugin_dir.joinpath(*tail) if tail else plugin_dir

        if base.is_dir():
            init = base / '__init__.py'
            if init.exists():
                return self._source_spec(name, fullname, init, True, [str(base)])
            # Regular package with no __init__.py: refuse a planted
            # __init__.<ext>; otherwise treat as a namespace package (no body).
            self._refuse_if_unverifiable_present(base, '__init__', name)
            return None

        source = base.with_suffix('.py')
        if source.exists():
            return self._source_spec(name, fullname, source, False, None)

        # No .py for this name: refuse a planted .so / sourceless .pyc, else defer
        # (a genuinely-absent module -> normal ModuleNotFoundError).
        self._refuse_if_unverifiable_present(base.parent, base.name, name)
        return None

    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)]