feat: plugin API, manifests and loader

plugins/syntax/dangerous_eval/dangerous_eval.py

@@ -0,0 +1,62 @@
+"""Report calls to the dynamic-code builtins, whatever name the file gives them."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import ClassVar
+
+from coretrace_python.analysis import AnyAnalysis
+from coretrace_python.findings import Confidence, Finding, Severity
+from coretrace_python.hir import nodes
+from coretrace_python.hir.visitors import Node, children
+from coretrace_python.plugins import Plugin, PluginContext
+from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeId, ScopeTable
+from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId
+
+DANGEROUS = frozenset({SymbolId("python.builtins.eval"), SymbolId("python.builtins.exec")})
+
+
+class DangerousEvalPlugin(Plugin):
+    name: ClassVar[str] = "dangerous-eval"
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({ScopeAnalysis, SymbolAnalysis})
+
+    def analyze(self, ctx: PluginContext) -> list[Finding]:
+        scopes = ctx.get(ScopeAnalysis)
+        symbols = ctx.get(SymbolAnalysis)
+        findings: list[Finding] = []
+        for function in ctx.functions():
+            for call, scope_id, enclosing in _calls(function, scopes, scopes.scope_for(function).id):
+                if not isinstance(call.callee, nodes.Name):
+                    continue
+                symbol = symbols.resolve(scope_id, call.callee.identifier)
+                if symbol in DANGEROUS:
+                    findings.append(
+                        Finding(
+                            rule_id="dangerous-eval",
+                            message=f"call to {symbol} executes dynamically built code",
+                            severity=Severity.HIGH,
+                            confidence=Confidence.HIGH,
+                            span=call.span,
+                            function=enclosing,
+                        )
+                    )
+        return findings
+
+
+def _calls(
+    function: nodes.Function, scopes: ScopeTable, scope_id: ScopeId
+) -> Iterator[tuple[nodes.Call, ScopeId, str]]:
+    """Yield every call in ``function`` with the scope it is evaluated in."""
+
+    def walk(node: Node, scope_id: ScopeId, enclosing: str) -> Iterator[tuple[nodes.Call, ScopeId, str]]:
+        if isinstance(node, nodes.Call):
+            yield node, scope_id, enclosing
+        if isinstance(node, nodes.Function):
+            scope_id, enclosing = scopes.scope_for(node).id, node.name
+        elif isinstance(node, nodes.Class | nodes.Comprehension):
+            scope_id = scopes.scope_for(node).id
+        for child in children(node):
+            yield from walk(child, scope_id, enclosing)
+
+    for statement in function.body:
+        yield from walk(statement, scope_id, function.name)

plugins/syntax/dangerous_eval/plugin.toml

@@ -0,0 +1,9 @@
+name = "dangerous-eval"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = ["semantic.scopes", "semantic.symbols"]
+provides = ["vulnerability.dangerous-eval"]
+
+[entrypoint]
+module = "dangerous_eval"
+class = "DangerousEvalPlugin"

src/coretrace_python/analysis/manager.py

@@ -54,6 +54,14 @@ def register(self, *analyses: AnyAnalysis) -> None:
             self._registry.add(analysis)
             self._check_acyclic(analysis)
 
+    def analysis(self, name: str) -> AnyAnalysis:
+        """Look a registered analysis up by its declared name."""
+
+        for analysis in self._registry:
+            if analysis.name == name:
+                return analysis
+        raise KeyError(name)
+
     def dependencies(self, analysis: AnyAnalysis) -> frozenset[AnyAnalysis]:
         """Transitive closure of ``analysis.requires``."""
 

src/coretrace_python/findings/__init__.py

@@ -0,0 +1,5 @@
+"""Normalized findings produced by plugins and consumed by reporters."""
+
+from coretrace_python.findings.model import FINDING_SCHEMA_VERSION, Confidence, Finding, Severity
+
+__all__ = ["FINDING_SCHEMA_VERSION", "Confidence", "Finding", "Severity"]

src/coretrace_python/findings/model.py

@@ -0,0 +1,49 @@
+"""Normalized findings (architecture §23).
+
+A finding references lightweight data only: a rule, a message, a source span and the
+name of the enclosing function. It never carries CFG or taint-path copies; those are
+reconstructed at report time.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from enum import Enum
+from types import MappingProxyType
+
+from coretrace_python.source import SourceSpan
+
+FINDING_SCHEMA_VERSION = 1
+
+
+class Severity(Enum):
+    INFO = "info"
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class Confidence(Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+def _no_metadata() -> Mapping[str, str]:
+    return MappingProxyType({})
+
+
+@dataclass(frozen=True)
+class Finding:
+    rule_id: str
+    message: str
+    severity: Severity
+    confidence: Confidence
+    span: SourceSpan
+    function: str | None = None
+    metadata: Mapping[str, str] = field(default_factory=_no_metadata)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata)))

src/coretrace_python/hir/visitors.py

@@ -0,0 +1,27 @@
+"""Generic traversal helpers over PyHIR nodes."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from dataclasses import fields
+
+from coretrace_python.hir import nodes
+
+Node = nodes.Statement | nodes.Expression | nodes.ComprehensionGenerator | nodes.Keyword
+
+
+def children(node: Node) -> Iterator[Node]:
+    """Yield the direct child nodes of ``node`` in source order."""
+
+    for field in fields(node):
+        value = getattr(node, field.name)
+        if _is_node(value):
+            yield value
+        elif isinstance(value, tuple):
+            for item in value:
+                if _is_node(item):
+                    yield item
+
+
+def _is_node(value: object) -> bool:
+    return hasattr(value, "span") and not isinstance(value, nodes.Parameter | nodes.ImportAlias)

src/coretrace_python/plugins/__init__.py

@@ -0,0 +1,34 @@
+"""Plugin API, manifests, loader and registry (architecture §13, §14, §32–§34)."""
+
+from coretrace_python.plugins.api import PLUGIN_API_VERSION, Plugin, PluginContext, run_plugins
+from coretrace_python.plugins.loader import (
+    IncompatiblePluginError,
+    LoadedPlugin,
+    discover_plugins,
+    load_plugin,
+)
+from coretrace_python.plugins.manifest import (
+    Entrypoint,
+    ManifestError,
+    PluginManifest,
+    VersionRange,
+    load_manifest,
+)
+from coretrace_python.plugins.registry import PluginRegistry
+
+__all__ = [
+    "PLUGIN_API_VERSION",
+    "Entrypoint",
+    "IncompatiblePluginError",
+    "LoadedPlugin",
+    "ManifestError",
+    "Plugin",
+    "PluginContext",
+    "PluginManifest",
+    "PluginRegistry",
+    "VersionRange",
+    "discover_plugins",
+    "load_manifest",
+    "load_plugin",
+    "run_plugins",
+]

src/coretrace_python/plugins/api.py

@@ -0,0 +1,78 @@
+"""The typed plugin contract (architecture §13, §32, §33, §34).
+
+A plugin declares the analyses it needs and consumes them through a ``PluginContext``,
+a deliberately narrow view of the engine: the module, its functions and the declared
+analyses. Nothing else is reachable, which keeps plugins on the stable API and leaves
+room for out-of-process isolation later.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Iterable, Sequence
+from typing import Any, ClassVar, overload
+
+from coretrace_python.analysis import (
+    Analysis,
+    AnalysisManager,
+    AnyAnalysis,
+    FunctionAnalysis,
+    UndeclaredDependencyError,
+)
+from coretrace_python.analysis.provider import R
+from coretrace_python.findings import Finding
+from coretrace_python.hir import nodes
+
+PLUGIN_API_VERSION = 1
+
+
+class Plugin(ABC):
+    """A detector: declares ``requires`` and turns analysis results into findings."""
+
+    name: ClassVar[str]
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset()
+
+    @abstractmethod
+    def analyze(self, ctx: PluginContext) -> Sequence[Finding]:
+        raise NotImplementedError
+
+
+class PluginContext:
+    """What one plugin may see while analysing one module."""
+
+    def __init__(self, manager: AnalysisManager, plugin: Plugin) -> None:
+        self._manager = manager
+        self._plugin = plugin
+
+    @property
+    def module(self) -> nodes.Module:
+        return self._manager.module
+
+    def functions(self) -> tuple[nodes.Function, ...]:
+        return tuple(s for s in self.module.body if isinstance(s, nodes.Function))
+
+    @overload
+    def get(self, analysis: type[Analysis[R]], function: None = None) -> R: ...
+
+    @overload
+    def get(self, analysis: type[FunctionAnalysis[R]], function: nodes.Function) -> R: ...
+
+    def get(self, analysis: AnyAnalysis, function: nodes.Function | None = None) -> Any:
+        if analysis not in self._plugin.requires:
+            raise UndeclaredDependencyError(
+                f"{self._plugin.name} requests {analysis.name} without declaring it in requires"
+            )
+        if function is None:
+            assert issubclass(analysis, Analysis)
+            return self._manager.get(analysis)
+        assert issubclass(analysis, FunctionAnalysis)
+        return self._manager.get(analysis, function)
+
+
+def run_plugins(manager: AnalysisManager, plugins: Iterable[Plugin]) -> tuple[Finding, ...]:
+    """Run every plugin against one shared manager and concatenate their findings."""
+
+    findings: list[Finding] = []
+    for plugin in plugins:
+        findings.extend(plugin.analyze(PluginContext(manager, plugin)))
+    return tuple(findings)

src/coretrace_python/plugins/loader.py

@@ -0,0 +1,84 @@
+"""In-process plugin loading (architecture §14, §34)."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+from dataclasses import dataclass
+from pathlib import Path
+
+from coretrace_python.analysis import AnalysisManager
+from coretrace_python.plugins.api import PLUGIN_API_VERSION, Plugin
+from coretrace_python.plugins.manifest import ManifestError, PluginManifest, load_manifest
+
+MANIFEST_FILENAME = "plugin.toml"
+
+
+class IncompatiblePluginError(Exception):
+    """The plugin targets a plugin API version this engine does not provide."""
+
+
+@dataclass(frozen=True)
+class LoadedPlugin:
+    manifest: PluginManifest
+    plugin: Plugin
+    directory: Path
+
+
+def load_plugin(directory: Path, manager: AnalysisManager) -> LoadedPlugin:
+    """Load the plugin in ``directory`` and check it against its manifest."""
+
+    manifest_path = directory / MANIFEST_FILENAME
+    manifest = load_manifest(manifest_path)
+    if not manifest.plugin_api.contains(PLUGIN_API_VERSION):
+        raise IncompatiblePluginError(
+            f"plugin {manifest.name!r} requires plugin_api {manifest.plugin_api}"
+            f" but this engine provides {PLUGIN_API_VERSION}"
+        )
+    for analysis_name in manifest.requires:
+        try:
+            manager.analysis(analysis_name)
+        except KeyError:
+            raise ManifestError(
+                f"{manifest_path}: unknown analysis {analysis_name!r} in requires"
+            ) from None
+
+    plugin_class = _import_entrypoint(directory, manifest)
+    declared = {analysis.name for analysis in plugin_class.requires}
+    if declared != set(manifest.requires):
+        raise ManifestError(
+            f"{manifest_path}: manifest requires {sorted(manifest.requires)}"
+            f" but {manifest.entrypoint.class_name} declares {sorted(declared)}"
+        )
+    return LoadedPlugin(manifest, plugin_class(), directory)
+
+
+def discover_plugins(root: Path, manager: AnalysisManager) -> tuple[LoadedPlugin, ...]:
+    """Load every plugin below ``root``, in path order."""
+
+    return tuple(
+        load_plugin(manifest_path.parent, manager)
+        for manifest_path in sorted(root.rglob(MANIFEST_FILENAME))
+    )
+
+
+def _import_entrypoint(directory: Path, manifest: PluginManifest) -> type[Plugin]:
+    module_path = directory / f"{manifest.entrypoint.module}.py"
+    if not module_path.is_file():
+        raise ManifestError(
+            f"{directory / MANIFEST_FILENAME}: entrypoint module"
+            f" {manifest.entrypoint.module!r} not found"
+        )
+    unique = hashlib.sha1(str(module_path.resolve()).encode()).hexdigest()[:12]
+    spec = importlib.util.spec_from_file_location(f"coretrace_plugin_{unique}", module_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+
+    candidate = getattr(module, manifest.entrypoint.class_name, None)
+    if not (isinstance(candidate, type) and issubclass(candidate, Plugin)):
+        raise ManifestError(
+            f"{directory / MANIFEST_FILENAME}: entrypoint class"
+            f" {manifest.entrypoint.class_name!r} is not a Plugin"
+        )
+    return candidate