From 1b8d51bcf9e66c5f2b035a6e22c3eb7afb5a57de Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 06:29:02 +0200 Subject: [PATCH 1/2] test: define plugin API, manifest and loader behavior Acceptance tests for typed plugins with declared requires, a narrow plugin context, lightweight findings, TOML manifests with plugin_api ranges, an in-process loader and registry, and the dangerous_eval reference plugin. Co-Authored-By: Claude Fable 5.1 --- tests/test_plugins.py | 431 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 431 insertions(+) create mode 100644 tests/test_plugins.py diff --git a/tests/test_plugins.py b/tests/test_plugins.py new file mode 100644 index 0000000..fcb1f95 --- /dev/null +++ b/tests/test_plugins.py @@ -0,0 +1,431 @@ +"""Acceptance tests for the plugin API, manifests, loader and registry. + +``docs/architecture.md`` §13 Plugin API, §14 Plugin Manifests, §23 Findings, §32 Stable +API, §33 Versioning, §34 Plugin Isolation. + +Contract under test: + +- ``Plugin`` subclasses declare ``name`` and ``requires`` and implement + ``analyze(ctx) -> Sequence[Finding]``. The context only serves declared analyses. +- ``Finding`` is an immutable record with lightweight fields, never a graph copy. +- A plugin directory holds ``plugin.toml`` (name, version, plugin_api, requires, + provides, entrypoint) next to its module. The loader checks the API range, the + entrypoint and that manifest ``requires`` match the class. +- The reference ``dangerous_eval`` plugin under ``plugins/syntax`` runs end to end. + +Expected to remain red until ``coretrace_python.plugins`` and +``coretrace_python.findings`` exist. +""" + +from __future__ import annotations + +import dataclasses +from pathlib import Path +from typing import ClassVar + +import pytest + +from coretrace_python.analysis import AnalysisManager, UndeclaredDependencyError +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.semantic import SEMANTIC_ANALYSES +from coretrace_python.semantic.scopes import ScopeAnalysis +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId +from coretrace_python.source import SourceId, SourceManager, SourceSpan + +try: + from coretrace_python.findings import FINDING_SCHEMA_VERSION, Confidence, Finding, Severity + from coretrace_python.plugins import ( + PLUGIN_API_VERSION, + IncompatiblePluginError, + LoadedPlugin, + ManifestError, + Plugin, + PluginContext, + PluginManifest, + PluginRegistry, + VersionRange, + discover_plugins, + load_manifest, + load_plugin, + run_plugins, + ) +except ImportError as error: # pragma: no cover - red until the plugin API lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_plugin_api() -> None: + if MISSING is not None: + pytest.fail(f"plugin API is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +DANGEROUS_EVAL = REPO / "plugins" / "syntax" / "dangerous_eval" + + +def manager_for(source_text: str) -> AnalysisManager: + module = build_hir(SourceManager().add_source("plugged.py", source_text)) + manager = AnalysisManager(module) + manager.register(*SEMANTIC_ANALYSES) + return manager + + +def span(line: int = 1, column: int = 1) -> SourceSpan: + return SourceSpan(SourceId("plugged.py"), line, column) + + +# --------------------------------------------------------------------------- stub plugins + + +if MISSING is None: # the stubs subclass the base class under test + + class CountFunctions(Plugin): + name: ClassVar[str] = "test.count-functions" + + def analyze(self, ctx: PluginContext) -> list[Finding]: + return [ + Finding( + rule_id="test.function", + message=f"function {function.name}", + severity=Severity.INFO, + confidence=Confidence.HIGH, + span=function.span, + function=function.name, + ) + for function in ctx.functions() + ] + + class UsesSymbols(Plugin): + name: ClassVar[str] = "test.uses-symbols" + requires: ClassVar[frozenset[type[object]]] = frozenset({ScopeAnalysis, SymbolAnalysis}) + + def analyze(self, ctx: PluginContext) -> list[Finding]: + scopes = ctx.get(ScopeAnalysis) + symbols = ctx.get(SymbolAnalysis) + findings = [] + for function in ctx.functions(): + symbol = symbols.resolve(scopes.scope_for(function).id, "os") + if symbol == SymbolId("python.os"): + findings.append( + Finding("test.os", "uses os", Severity.LOW, Confidence.MEDIUM, function.span) + ) + return findings + + class Sneaky(Plugin): + """Requests an analysis it did not declare.""" + + name: ClassVar[str] = "test.sneaky" + + def analyze(self, ctx: PluginContext) -> list[Finding]: + ctx.get(SymbolAnalysis) + return [] + + +# --------------------------------------------------------------------------- findings + + +def test_findings_are_immutable_lightweight_records() -> None: + finding = Finding("rule", "message", Severity.HIGH, Confidence.HIGH, span(2, 5)) + + assert finding.function is None + assert dict(finding.metadata) == {} + with pytest.raises(dataclasses.FrozenInstanceError): + finding.message = "other" # type: ignore[misc] + with pytest.raises(TypeError): + finding.metadata["k"] = "v" # type: ignore[index] + + +def test_versions_are_declared() -> None: + assert PLUGIN_API_VERSION == 1 + assert FINDING_SCHEMA_VERSION >= 1 + + +# --------------------------------------------------------------------------- plugin contract + + +def test_plugins_receive_the_module_and_its_functions() -> None: + manager = manager_for("def a():\n pass\n\ndef b():\n pass\n") + + findings = run_plugins(manager, [CountFunctions()]) + + assert [f.function for f in findings] == ["a", "b"] + assert findings[0].rule_id == "test.function" + assert findings[0].span.start_line == 1 + assert findings[1].span.start_line == 4 + + +def test_plugins_consume_declared_analyses() -> None: + manager = manager_for("import os\n\ndef run(c):\n os.system(c)\n\ndef idle():\n pass\n") + + findings = run_plugins(manager, [UsesSymbols()]) + + assert [f.rule_id for f in findings] == ["test.os"] + assert findings[0].span.start_line == 3 + + +def test_undeclared_analysis_requests_are_rejected() -> None: + manager = manager_for("def a():\n pass\n") + + with pytest.raises(UndeclaredDependencyError, match="test.sneaky.*semantic.symbols"): + run_plugins(manager, [Sneaky()]) + + +def test_analyses_are_shared_between_plugins() -> None: + manager = manager_for("import os\n\ndef run(c):\n os.system(c)\n") + + run_plugins(manager, [UsesSymbols(), UsesSymbols()]) + + assert manager.is_cached(SymbolAnalysis) + assert manager.get(SymbolAnalysis) is manager.get(SymbolAnalysis) + + +def test_findings_are_returned_in_plugin_order() -> None: + manager = manager_for("import os\n\ndef run(c):\n os.system(c)\n") + + findings = run_plugins(manager, [UsesSymbols(), CountFunctions()]) + + assert [f.rule_id for f in findings] == ["test.os", "test.function"] + + +# --------------------------------------------------------------------------- version ranges + + +@pytest.mark.parametrize( + "spec, inside, outside", + [ + (">=1,<2", [1], [0, 2]), + ("1", [1], [0, 2]), + ("==1", [1], [2]), + (">=1", [1, 5], [0]), + ], +) +def test_version_ranges(spec: str, inside: list[int], outside: list[int]) -> None: + version_range = VersionRange.parse(spec) + assert all(version_range.contains(v) for v in inside) + assert not any(version_range.contains(v) for v in outside) + + +@pytest.mark.parametrize("spec", ["", "~1", ">=", "1.0", ">=1,,<2"]) +def test_invalid_version_ranges_are_rejected(spec: str) -> None: + with pytest.raises(ManifestError): + VersionRange.parse(spec) + + +# --------------------------------------------------------------------------- manifests + + +def test_reference_manifest_is_parsed() -> None: + manifest = load_manifest(DANGEROUS_EVAL / "plugin.toml") + + assert isinstance(manifest, PluginManifest) + assert manifest.name == "dangerous-eval" + assert manifest.version == "1.0.0" + assert manifest.plugin_api.contains(PLUGIN_API_VERSION) + assert manifest.requires == ("semantic.scopes", "semantic.symbols") + assert manifest.provides == ("vulnerability.dangerous-eval",) + assert manifest.entrypoint.module == "dangerous_eval" + assert manifest.entrypoint.class_name == "DangerousEvalPlugin" + + +def write_manifest(directory: Path, **overrides: object) -> Path: + fields: dict[str, object] = { + "name": "sample", + "version": "0.1.0", + "plugin_api": ">=1,<2", + "requires": [], + "provides": ["test.sample"], + } + fields.update({k: v for k, v in overrides.items() if k != "entrypoint"}) + entrypoint = overrides.get("entrypoint", {"module": "sample", "class": "SamplePlugin"}) + lines = [] + for key, value in fields.items(): + if isinstance(value, list): + lines.append(f"{key} = [{', '.join(repr(v) for v in value)}]") + else: + lines.append(f"{key} = {value!r}") + if entrypoint is not None: + assert isinstance(entrypoint, dict) + lines.append("[entrypoint]") + lines.extend(f"{k} = {v!r}" for k, v in entrypoint.items()) + path = directory / "plugin.toml" + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return path + + +def write_plugin(directory: Path, requires: str = "frozenset()") -> None: + (directory / "sample.py").write_text( + "from typing import ClassVar\n\n" + "from coretrace_python.plugins import Plugin, PluginContext\n" + "from coretrace_python.semantic.scopes import ScopeAnalysis\n" + "from coretrace_python.semantic.symbols import SymbolAnalysis\n\n\n" + "class SamplePlugin(Plugin):\n" + ' name: ClassVar[str] = "sample"\n' + f" requires = {requires}\n\n" + " def analyze(self, ctx: PluginContext) -> list:\n" + " return []\n\n\n" + "class NotAPlugin:\n" + " pass\n", + encoding="utf-8", + ) + + +@pytest.mark.parametrize( + "missing", ["name", "version", "plugin_api", "requires", "provides", "entrypoint"] +) +def test_manifest_requires_every_field(tmp_path: Path, missing: str) -> None: + path = write_manifest(tmp_path, **{missing: None}) + if missing != "entrypoint": + text = "\n".join( + line for line in path.read_text().splitlines() if not line.startswith(f"{missing} =") + ) + path.write_text(text + "\n", encoding="utf-8") + + with pytest.raises(ManifestError, match=missing): + load_manifest(path) + + +# --------------------------------------------------------------------------- loader + + +def test_loads_the_reference_plugin() -> None: + loaded = load_plugin(DANGEROUS_EVAL, manager_for("")) + + assert isinstance(loaded, LoadedPlugin) + assert loaded.manifest.name == "dangerous-eval" + assert isinstance(loaded.plugin, Plugin) + assert loaded.plugin.name == "dangerous-eval" + assert loaded.plugin.requires == frozenset({ScopeAnalysis, SymbolAnalysis}) + + +def test_incompatible_plugin_api_is_rejected(tmp_path: Path) -> None: + write_manifest(tmp_path, plugin_api=">=2,<3") + write_plugin(tmp_path) + + with pytest.raises(IncompatiblePluginError, match="sample.*>=2,<3"): + load_plugin(tmp_path, manager_for("")) + + +def test_manifest_requires_must_name_registered_analyses(tmp_path: Path) -> None: + write_manifest(tmp_path, requires=["analysis.taint"]) + write_plugin(tmp_path) + + with pytest.raises(ManifestError, match="analysis.taint"): + load_plugin(tmp_path, manager_for("")) + + +def test_manifest_requires_must_match_the_class(tmp_path: Path) -> None: + write_manifest(tmp_path, requires=["semantic.scopes"]) + write_plugin(tmp_path, requires="frozenset({ScopeAnalysis, SymbolAnalysis})") + + with pytest.raises(ManifestError, match="semantic.symbols"): + load_plugin(tmp_path, manager_for("")) + + +def test_entrypoint_must_be_a_plugin(tmp_path: Path) -> None: + write_manifest(tmp_path, entrypoint={"module": "sample", "class": "NotAPlugin"}) + write_plugin(tmp_path) + + with pytest.raises(ManifestError, match="NotAPlugin"): + load_plugin(tmp_path, manager_for("")) + + +def test_missing_entrypoint_module_is_reported(tmp_path: Path) -> None: + write_manifest(tmp_path, entrypoint={"module": "absent", "class": "SamplePlugin"}) + + with pytest.raises(ManifestError, match="absent"): + load_plugin(tmp_path, manager_for("")) + + +def test_discovers_plugins_recursively_in_sorted_order(tmp_path: Path) -> None: + for name in ("b/second", "a/first"): + directory = tmp_path / name + directory.mkdir(parents=True) + write_manifest(directory, name=name.split("/")[1]) + write_plugin(directory) + + loaded = discover_plugins(tmp_path, manager_for("")) + + assert [plugin.manifest.name for plugin in loaded] == ["first", "second"] + + +# --------------------------------------------------------------------------- registry + + +def test_registry_indexes_plugins_by_name_and_capability() -> None: + loaded = load_plugin(DANGEROUS_EVAL, manager_for("")) + registry = PluginRegistry() + + registry.add(loaded) + + assert registry.plugin("dangerous-eval") is loaded + assert registry.providers("vulnerability.dangerous-eval") == (loaded,) + assert registry.providers("vulnerability.sql-injection") == () + assert [plugin.manifest.name for plugin in registry] == ["dangerous-eval"] + + +def test_registry_rejects_duplicate_names() -> None: + registry = PluginRegistry() + registry.add(load_plugin(DANGEROUS_EVAL, manager_for(""))) + + with pytest.raises(ManifestError, match="dangerous-eval"): + registry.add(load_plugin(DANGEROUS_EVAL, manager_for(""))) + + +# --------------------------------------------------------------------------- reference plugin + + +def test_dangerous_eval_reports_builtin_eval_calls() -> None: + manager = manager_for("def run(code):\n eval(code)\n return exec(code)\n") + loaded = load_plugin(DANGEROUS_EVAL, manager) + + findings = run_plugins(manager, [loaded.plugin]) + + assert [(f.rule_id, f.span.start_line, f.function) for f in findings] == [ + ("dangerous-eval", 2, "run"), + ("dangerous-eval", 3, "run"), + ] + assert findings[0].severity is Severity.HIGH + assert "python.builtins.eval" in findings[0].message + assert "python.builtins.exec" in findings[1].message + + +def test_dangerous_eval_follows_symbols_not_names() -> None: + manager = manager_for( + "from ast import literal_eval as eval\n\n" + "def parse(text):\n" + " return eval(text)\n\n" + "def shadowed(eval, text):\n" + " return eval(text)\n" + ) + loaded = load_plugin(DANGEROUS_EVAL, manager) + + assert run_plugins(manager, [loaded.plugin]) == () + + +def test_dangerous_eval_ignores_modules_without_functions() -> None: + manager = manager_for("x = 1\n") + loaded = load_plugin(DANGEROUS_EVAL, manager) + + assert run_plugins(manager, [loaded.plugin]) == () + + +def test_plugin_context_is_a_narrow_view() -> None: + manager = manager_for("def a():\n pass\n") + seen: list[object] = [] + + class Inspect(Plugin): + name: ClassVar[str] = "test.inspect" + + def analyze(self, ctx: PluginContext) -> list[Finding]: + seen.append(ctx) + return [] + + run_plugins(manager, [Inspect()]) + + ctx = seen[0] + assert isinstance(ctx.module, nodes.Module) + assert not hasattr(ctx, "register") + assert not hasattr(ctx, "run") From a12f9f5334dfb1d11892d838728bdf960c6d5449 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 06:31:28 +0200 Subject: [PATCH 2/2] feat: add plugin API, manifests, loader and registry Plugins declare requires and analyze a narrow PluginContext that only serves declared analyses. Findings are immutable lightweight records. Manifests are plugin.toml files with a plugin_api compatibility range, checked by an in-process loader against the entrypoint class and the registered analyses. The dangerous_eval reference plugin resolves eval and exec through SymbolAnalysis, so aliases and shadowing do not fool it. Co-Authored-By: Claude Fable 5.1 --- .../syntax/dangerous_eval/dangerous_eval.py | 62 ++++++++++ plugins/syntax/dangerous_eval/plugin.toml | 9 ++ src/coretrace_python/analysis/manager.py | 8 ++ src/coretrace_python/findings/__init__.py | 5 + src/coretrace_python/findings/model.py | 49 ++++++++ src/coretrace_python/hir/visitors.py | 27 +++++ src/coretrace_python/plugins/__init__.py | 34 ++++++ src/coretrace_python/plugins/api.py | 78 ++++++++++++ src/coretrace_python/plugins/loader.py | 84 +++++++++++++ src/coretrace_python/plugins/manifest.py | 112 ++++++++++++++++++ src/coretrace_python/plugins/registry.py | 34 ++++++ tests/test_plugins.py | 2 +- 12 files changed, 503 insertions(+), 1 deletion(-) create mode 100644 plugins/syntax/dangerous_eval/dangerous_eval.py create mode 100644 plugins/syntax/dangerous_eval/plugin.toml create mode 100644 src/coretrace_python/findings/__init__.py create mode 100644 src/coretrace_python/findings/model.py create mode 100644 src/coretrace_python/hir/visitors.py create mode 100644 src/coretrace_python/plugins/__init__.py create mode 100644 src/coretrace_python/plugins/api.py create mode 100644 src/coretrace_python/plugins/loader.py create mode 100644 src/coretrace_python/plugins/manifest.py create mode 100644 src/coretrace_python/plugins/registry.py diff --git a/plugins/syntax/dangerous_eval/dangerous_eval.py b/plugins/syntax/dangerous_eval/dangerous_eval.py new file mode 100644 index 0000000..7c8140d --- /dev/null +++ b/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) diff --git a/plugins/syntax/dangerous_eval/plugin.toml b/plugins/syntax/dangerous_eval/plugin.toml new file mode 100644 index 0000000..d27e4e9 --- /dev/null +++ b/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" diff --git a/src/coretrace_python/analysis/manager.py b/src/coretrace_python/analysis/manager.py index b6bb30c..82976b5 100644 --- a/src/coretrace_python/analysis/manager.py +++ b/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``.""" diff --git a/src/coretrace_python/findings/__init__.py b/src/coretrace_python/findings/__init__.py new file mode 100644 index 0000000..4692b91 --- /dev/null +++ b/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"] diff --git a/src/coretrace_python/findings/model.py b/src/coretrace_python/findings/model.py new file mode 100644 index 0000000..c3a7d27 --- /dev/null +++ b/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))) diff --git a/src/coretrace_python/hir/visitors.py b/src/coretrace_python/hir/visitors.py new file mode 100644 index 0000000..a7d8b7c --- /dev/null +++ b/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) diff --git a/src/coretrace_python/plugins/__init__.py b/src/coretrace_python/plugins/__init__.py new file mode 100644 index 0000000..6849504 --- /dev/null +++ b/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", +] diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py new file mode 100644 index 0000000..1ab6a71 --- /dev/null +++ b/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) diff --git a/src/coretrace_python/plugins/loader.py b/src/coretrace_python/plugins/loader.py new file mode 100644 index 0000000..cfba3c1 --- /dev/null +++ b/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 diff --git a/src/coretrace_python/plugins/manifest.py b/src/coretrace_python/plugins/manifest.py new file mode 100644 index 0000000..67f02c5 --- /dev/null +++ b/src/coretrace_python/plugins/manifest.py @@ -0,0 +1,112 @@ +"""Plugin manifests (architecture §14, §33). + +A manifest is a ``plugin.toml`` file next to the plugin module:: + + name = "sql-injection" + version = "1.0.0" + plugin_api = ">=1,<2" + requires = ["semantic.symbols", "analysis.taint"] + provides = ["vulnerability.sql-injection"] + + [entrypoint] + module = "sql_injection" + class = "SQLInjectionPlugin" +""" + +from __future__ import annotations + +import re +import tomllib +from dataclasses import dataclass +from pathlib import Path + + +class ManifestError(Exception): + """A manifest that cannot be read, or that disagrees with its plugin.""" + + +_CONSTRAINT = re.compile(r"^(>=|<=|==|>|<)?(\d+)$") +_COMPARE = { + "==": lambda actual, wanted: actual == wanted, + ">=": lambda actual, wanted: actual >= wanted, + "<=": lambda actual, wanted: actual <= wanted, + ">": lambda actual, wanted: actual > wanted, + "<": lambda actual, wanted: actual < wanted, +} + + +@dataclass(frozen=True) +class VersionRange: + """Integer API version constraints such as ``">=1,<2"`` or ``"1"``.""" + + spec: str + constraints: tuple[tuple[str, int], ...] + + @classmethod + def parse(cls, spec: str) -> VersionRange: + constraints: list[tuple[str, int]] = [] + for part in spec.split(","): + match = _CONSTRAINT.match(part.strip()) + if match is None: + raise ManifestError(f"invalid plugin_api range: {spec!r}") + operator, number = match.groups() + constraints.append((operator or "==", int(number))) + return cls(spec, tuple(constraints)) + + def contains(self, version: int) -> bool: + return all(_COMPARE[operator](version, wanted) for operator, wanted in self.constraints) + + def __str__(self) -> str: + return self.spec + + +@dataclass(frozen=True) +class Entrypoint: + module: str + class_name: str + + +@dataclass(frozen=True) +class PluginManifest: + name: str + version: str + plugin_api: VersionRange + requires: tuple[str, ...] + provides: tuple[str, ...] + entrypoint: Entrypoint + + +def load_manifest(path: Path) -> PluginManifest: + try: + data = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as error: + raise ManifestError(f"{path}: {error}") from error + + def string(key: str) -> str: + value = data.get(key) + if not isinstance(value, str) or not value: + raise ManifestError(f"{path}: missing or invalid field '{key}'") + return value + + def strings(key: str) -> tuple[str, ...]: + value = data.get(key) + if not isinstance(value, list) or not all(isinstance(v, str) for v in value): + raise ManifestError(f"{path}: missing or invalid field '{key}'") + return tuple(value) + + entrypoint = data.get("entrypoint") + if not isinstance(entrypoint, dict): + raise ManifestError(f"{path}: missing or invalid field 'entrypoint'") + module = entrypoint.get("module") + class_name = entrypoint.get("class") + if not isinstance(module, str) or not isinstance(class_name, str): + raise ManifestError(f"{path}: entrypoint needs 'module' and 'class'") + + return PluginManifest( + name=string("name"), + version=string("version"), + plugin_api=VersionRange.parse(string("plugin_api")), + requires=strings("requires"), + provides=strings("provides"), + entrypoint=Entrypoint(module, class_name), + ) diff --git a/src/coretrace_python/plugins/registry.py b/src/coretrace_python/plugins/registry.py new file mode 100644 index 0000000..6c2c203 --- /dev/null +++ b/src/coretrace_python/plugins/registry.py @@ -0,0 +1,34 @@ +"""Registry of loaded plugins, indexed by name and by provided capability.""" + +from __future__ import annotations + +from collections.abc import Iterator + +from coretrace_python.plugins.loader import LoadedPlugin +from coretrace_python.plugins.manifest import ManifestError + + +class PluginRegistry: + def __init__(self) -> None: + self._by_name: dict[str, LoadedPlugin] = {} + self._by_capability: dict[str, list[LoadedPlugin]] = {} + + def add(self, loaded: LoadedPlugin) -> None: + name = loaded.manifest.name + if name in self._by_name: + raise ManifestError(f"plugin {name!r} is already registered") + self._by_name[name] = loaded + for capability in loaded.manifest.provides: + self._by_capability.setdefault(capability, []).append(loaded) + + def plugin(self, name: str) -> LoadedPlugin: + return self._by_name[name] + + def providers(self, capability: str) -> tuple[LoadedPlugin, ...]: + return tuple(self._by_capability.get(capability, ())) + + def __iter__(self) -> Iterator[LoadedPlugin]: + return iter(self._by_name.values()) + + def __len__(self) -> int: + return len(self._by_name) diff --git a/tests/test_plugins.py b/tests/test_plugins.py index fcb1f95..27b6ea7 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -158,7 +158,7 @@ def test_plugins_receive_the_module_and_its_functions() -> None: def test_plugins_consume_declared_analyses() -> None: - manager = manager_for("import os\n\ndef run(c):\n os.system(c)\n\ndef idle():\n pass\n") + manager = manager_for("import os\n\ndef run(c):\n os.system(c)\n\ndef idle(os):\n pass\n") findings = run_plugins(manager, [UsesSymbols()])