diff --git a/README.md b/README.md index 19afc9d..2ccb077 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,12 @@ coretrace-python-analyzer --check app.py --plugins plugins/ --format sarif > rep ``` `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be -repeated. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing +repeated. The repository ships a first set under `plugins/`: standard-library security +models (`models/python_stdlib`), taint detectors for SQL injection, command injection, +path traversal, SSRF and XSS (`security/`), and syntactic detectors for `eval`/`exec` +and weak hashes (`syntax/`). Model plugins contribute sources, sinks and sanitizers; +detectors consume the shared taint result, so adding a framework model makes every +detector aware of it. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing was found, 1 when findings were reported, and 2 on a usage or analysis error. ## Emit PyIR diff --git a/plugins/models/python_stdlib/plugin.toml b/plugins/models/python_stdlib/plugin.toml new file mode 100644 index 0000000..78cdd04 --- /dev/null +++ b/plugins/models/python_stdlib/plugin.toml @@ -0,0 +1,9 @@ +name = "python-stdlib-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.python-stdlib"] + +[entrypoint] +module = "python_stdlib" +class = "PythonStdlibModels" diff --git a/plugins/models/python_stdlib/python_stdlib.py b/plugins/models/python_stdlib/python_stdlib.py new file mode 100644 index 0000000..9266bac --- /dev/null +++ b/plugins/models/python_stdlib/python_stdlib.py @@ -0,0 +1,41 @@ +"""Security models for the Python standard library (architecture §16).""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import Model, Sanitizer, Sink, Source, TaintKind + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class PythonStdlibModels(ModelPlugin): + name: ClassVar[str] = "python-stdlib-models" + models: ClassVar[tuple[Model, ...]] = ( + Source(_sym("builtins.input"), "stdin"), + Source(_sym("sys.stdin"), "stdin"), + Source(_sym("sys.argv"), "argv"), + Source(_sym("os.environ"), "environment"), + Sink(_sym("os.system"), TaintKind.COMMAND), + Sink(_sym("os.popen"), TaintKind.COMMAND), + Sink(_sym("subprocess.run"), TaintKind.COMMAND), + Sink(_sym("subprocess.call"), TaintKind.COMMAND), + Sink(_sym("subprocess.check_call"), TaintKind.COMMAND), + Sink(_sym("subprocess.check_output"), TaintKind.COMMAND), + Sink(_sym("subprocess.Popen"), TaintKind.COMMAND), + Sink(_sym("builtins.eval"), TaintKind.CODE), + Sink(_sym("builtins.exec"), TaintKind.CODE), + Sink(_sym("builtins.open"), TaintKind.PATH), + Sink(_sym("os.remove"), TaintKind.PATH), + Sink(_sym("os.unlink"), TaintKind.PATH), + Sink(_sym("os.rmdir"), TaintKind.PATH), + Sink(_sym("shutil.rmtree"), TaintKind.PATH), + Sink(_sym("urllib.request.urlopen"), TaintKind.SSRF), + Sanitizer(_sym("shlex.quote"), TaintKind.COMMAND), + Sanitizer(_sym("html.escape"), TaintKind.HTML), + Sanitizer(_sym("os.path.basename"), TaintKind.PATH), + ) diff --git a/plugins/security/command_injection/command_injection.py b/plugins/security/command_injection/command_injection.py new file mode 100644 index 0000000..f43942b --- /dev/null +++ b/plugins/security/command_injection/command_injection.py @@ -0,0 +1,17 @@ +"""Command injection: attacker-controlled input reaching a COMMAND sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class CommandInjectionPlugin(TaintDetector): + name: ClassVar[str] = "command-injection" + rule_id: ClassVar[str] = "command-injection" + kind: ClassVar[TaintKind] = TaintKind.COMMAND + severity: ClassVar[Severity] = Severity.HIGH + title: ClassVar[str] = "Command injection" diff --git a/plugins/security/command_injection/plugin.toml b/plugins/security/command_injection/plugin.toml new file mode 100644 index 0000000..e40b4d0 --- /dev/null +++ b/plugins/security/command_injection/plugin.toml @@ -0,0 +1,9 @@ +name = "command-injection" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows"] +provides = ["vulnerability.command-injection"] + +[entrypoint] +module = "command_injection" +class = "CommandInjectionPlugin" diff --git a/plugins/security/path_traversal/path_traversal.py b/plugins/security/path_traversal/path_traversal.py new file mode 100644 index 0000000..f43edbc --- /dev/null +++ b/plugins/security/path_traversal/path_traversal.py @@ -0,0 +1,17 @@ +"""Path traversal: attacker-controlled input reaching a PATH sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class PathTraversalPlugin(TaintDetector): + name: ClassVar[str] = "path-traversal" + rule_id: ClassVar[str] = "path-traversal" + kind: ClassVar[TaintKind] = TaintKind.PATH + severity: ClassVar[Severity] = Severity.HIGH + title: ClassVar[str] = "Path traversal" diff --git a/plugins/security/path_traversal/plugin.toml b/plugins/security/path_traversal/plugin.toml new file mode 100644 index 0000000..4936023 --- /dev/null +++ b/plugins/security/path_traversal/plugin.toml @@ -0,0 +1,9 @@ +name = "path-traversal" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows"] +provides = ["vulnerability.path-traversal"] + +[entrypoint] +module = "path_traversal" +class = "PathTraversalPlugin" diff --git a/plugins/security/sql_injection/plugin.toml b/plugins/security/sql_injection/plugin.toml new file mode 100644 index 0000000..c1a6587 --- /dev/null +++ b/plugins/security/sql_injection/plugin.toml @@ -0,0 +1,9 @@ +name = "sql-injection" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows"] +provides = ["vulnerability.sql-injection"] + +[entrypoint] +module = "sql_injection" +class = "SqlInjectionPlugin" diff --git a/plugins/security/sql_injection/sql_injection.py b/plugins/security/sql_injection/sql_injection.py new file mode 100644 index 0000000..d3b7179 --- /dev/null +++ b/plugins/security/sql_injection/sql_injection.py @@ -0,0 +1,17 @@ +"""SQL injection: attacker-controlled input reaching a SQL sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class SqlInjectionPlugin(TaintDetector): + name: ClassVar[str] = "sql-injection" + rule_id: ClassVar[str] = "sql-injection" + kind: ClassVar[TaintKind] = TaintKind.SQL + severity: ClassVar[Severity] = Severity.HIGH + title: ClassVar[str] = "SQL injection" diff --git a/plugins/security/ssrf/plugin.toml b/plugins/security/ssrf/plugin.toml new file mode 100644 index 0000000..084fcd3 --- /dev/null +++ b/plugins/security/ssrf/plugin.toml @@ -0,0 +1,9 @@ +name = "ssrf" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows"] +provides = ["vulnerability.ssrf"] + +[entrypoint] +module = "ssrf" +class = "SsrfPlugin" diff --git a/plugins/security/ssrf/ssrf.py b/plugins/security/ssrf/ssrf.py new file mode 100644 index 0000000..61b0788 --- /dev/null +++ b/plugins/security/ssrf/ssrf.py @@ -0,0 +1,17 @@ +"""Server-side request forgery: attacker-controlled input reaching a SSRF sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class SsrfPlugin(TaintDetector): + name: ClassVar[str] = "ssrf" + rule_id: ClassVar[str] = "ssrf" + kind: ClassVar[TaintKind] = TaintKind.SSRF + severity: ClassVar[Severity] = Severity.HIGH + title: ClassVar[str] = "Server-side request forgery" diff --git a/plugins/security/xss/plugin.toml b/plugins/security/xss/plugin.toml new file mode 100644 index 0000000..bad14cf --- /dev/null +++ b/plugins/security/xss/plugin.toml @@ -0,0 +1,9 @@ +name = "xss" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows"] +provides = ["vulnerability.xss"] + +[entrypoint] +module = "xss" +class = "XssPlugin" diff --git a/plugins/security/xss/xss.py b/plugins/security/xss/xss.py new file mode 100644 index 0000000..81aabd8 --- /dev/null +++ b/plugins/security/xss/xss.py @@ -0,0 +1,17 @@ +"""Cross-site scripting: attacker-controlled input reaching a HTML sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class XssPlugin(TaintDetector): + name: ClassVar[str] = "xss" + rule_id: ClassVar[str] = "xss" + kind: ClassVar[TaintKind] = TaintKind.HTML + severity: ClassVar[Severity] = Severity.HIGH + title: ClassVar[str] = "Cross-site scripting" diff --git a/plugins/syntax/dangerous_eval/dangerous_eval.py b/plugins/syntax/dangerous_eval/dangerous_eval.py index 7c8140d..7b428fa 100644 --- a/plugins/syntax/dangerous_eval/dangerous_eval.py +++ b/plugins/syntax/dangerous_eval/dangerous_eval.py @@ -2,61 +2,18 @@ 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 +from coretrace_python.findings import Severity +from coretrace_python.plugins import SymbolCallDetector +from coretrace_python.semantic.symbols import SymbolId -DANGEROUS = frozenset({SymbolId("python.builtins.eval"), SymbolId("python.builtins.exec")}) - -class DangerousEvalPlugin(Plugin): +class DangerousEvalPlugin(SymbolCallDetector): 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) + rule_id: ClassVar[str] = "dangerous-eval" + symbols: ClassVar[frozenset[SymbolId]] = frozenset( + {SymbolId("python.builtins.eval"), SymbolId("python.builtins.exec")} + ) + severity: ClassVar[Severity] = Severity.HIGH + message_template: ClassVar[str] = "call to {symbol} executes dynamically built code" diff --git a/plugins/syntax/dangerous_eval/plugin.toml b/plugins/syntax/dangerous_eval/plugin.toml index d27e4e9..4e0eb35 100644 --- a/plugins/syntax/dangerous_eval/plugin.toml +++ b/plugins/syntax/dangerous_eval/plugin.toml @@ -1,7 +1,7 @@ name = "dangerous-eval" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["semantic.scopes", "semantic.symbols"] +requires = ["ir.ssa"] provides = ["vulnerability.dangerous-eval"] [entrypoint] diff --git a/plugins/syntax/weak_crypto/plugin.toml b/plugins/syntax/weak_crypto/plugin.toml new file mode 100644 index 0000000..5395ac5 --- /dev/null +++ b/plugins/syntax/weak_crypto/plugin.toml @@ -0,0 +1,9 @@ +name = "weak-crypto" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["ir.ssa"] +provides = ["vulnerability.weak-crypto"] + +[entrypoint] +module = "weak_crypto" +class = "WeakCryptoPlugin" diff --git a/plugins/syntax/weak_crypto/weak_crypto.py b/plugins/syntax/weak_crypto/weak_crypto.py new file mode 100644 index 0000000..30e365b --- /dev/null +++ b/plugins/syntax/weak_crypto/weak_crypto.py @@ -0,0 +1,19 @@ +"""Report uses of broken hash algorithms.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import SymbolCallDetector +from coretrace_python.semantic.symbols import SymbolId + + +class WeakCryptoPlugin(SymbolCallDetector): + name: ClassVar[str] = "weak-crypto" + rule_id: ClassVar[str] = "weak-crypto" + symbols: ClassVar[frozenset[SymbolId]] = frozenset( + {SymbolId("python.hashlib.md5"), SymbolId("python.hashlib.sha1")} + ) + severity: ClassVar[Severity] = Severity.MEDIUM + message_template: ClassVar[str] = "call to {symbol} uses a broken hash algorithm" diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 6c42f84..9c46024 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -42,25 +42,40 @@ ) -def build_manager(module: nodes.Module, models: SecurityModelRegistry | None = None) -> AnalysisManager: +def build_manager( + module: nodes.Module, models: SecurityModelRegistry | None = None +) -> AnalysisManager: """A manager with every engine analysis registered and the security models provided.""" - manager = AnalysisManager(module) - manager.register(*ALL_ANALYSES) + manager = _register_all(module) manager.provide(SecurityModelAnalysis, (models or SecurityModelRegistry()).freeze()) return manager def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ...]: - """Run every plugin found under ``plugin_roots`` against ``source``.""" + """Run every plugin found under ``plugin_roots`` against ``source``. - manager = build_manager(build_hir(source)) + Plugins are loaded first so the models they contribute are all registered before the + table is provided; framework models and detectors compose without knowing each other. + """ + + manager = _register_all(build_hir(source)) registry = PluginRegistry() for root in plugin_roots: for loaded in discover_plugins(root, manager): registry.add(loaded) + models = SecurityModelRegistry() + for loaded in registry: + models.register(*loaded.plugin.models) + manager.provide(SecurityModelAnalysis, models.freeze()) return run_plugins(manager, [loaded.plugin for loaded in registry]) +def _register_all(module: nodes.Module) -> AnalysisManager: + manager = AnalysisManager(module) + manager.register(*ALL_ANALYSES) + return manager + + def report(findings: Sequence[Finding]) -> Report: return Report(tuple(findings), TOOL_NAME, __version__) diff --git a/src/coretrace_python/plugins/__init__.py b/src/coretrace_python/plugins/__init__.py index 6849504..2611e64 100644 --- a/src/coretrace_python/plugins/__init__.py +++ b/src/coretrace_python/plugins/__init__.py @@ -1,6 +1,13 @@ """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.api import ( + PLUGIN_API_VERSION, + ModelPlugin, + Plugin, + PluginContext, + run_plugins, +) +from coretrace_python.plugins.detectors import SymbolCallDetector, TaintDetector from coretrace_python.plugins.loader import ( IncompatiblePluginError, LoadedPlugin, @@ -22,10 +29,13 @@ "IncompatiblePluginError", "LoadedPlugin", "ManifestError", + "ModelPlugin", "Plugin", "PluginContext", "PluginManifest", "PluginRegistry", + "SymbolCallDetector", + "TaintDetector", "VersionRange", "discover_plugins", "load_manifest", diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 1ab6a71..60b2b4c 100644 --- a/src/coretrace_python/plugins/api.py +++ b/src/coretrace_python/plugins/api.py @@ -22,21 +22,31 @@ from coretrace_python.analysis.provider import R from coretrace_python.findings import Finding from coretrace_python.hir import nodes +from coretrace_python.taint import Model PLUGIN_API_VERSION = 1 class Plugin(ABC): - """A detector: declares ``requires`` and turns analysis results into findings.""" + """Declares the analyses it needs and the security models it contributes, + then turns analysis results into findings.""" name: ClassVar[str] requires: ClassVar[frozenset[AnyAnalysis]] = frozenset() + models: ClassVar[tuple[Model, ...]] = () @abstractmethod def analyze(self, ctx: PluginContext) -> Sequence[Finding]: raise NotImplementedError +class ModelPlugin(Plugin): + """A plugin that only contributes security models (architecture §15 providers).""" + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + return () + + class PluginContext: """What one plugin may see while analysing one module.""" diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py new file mode 100644 index 0000000..cfc8799 --- /dev/null +++ b/src/coretrace_python/plugins/detectors.py @@ -0,0 +1,93 @@ +"""Reusable detector bases (architecture §15). + +``TaintDetector`` turns the flows of one taint kind into findings; every security +detector is a few class attributes on top of it. ``SymbolCallDetector`` reports calls +whose callee resolves to a canonical symbol, through imports and local aliases alike, +by reading the SSA form. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.ir.model import Call, Symbol +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins.api import Plugin, PluginContext +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import TaintAnalysis, TaintKind + + +class TaintDetector(Plugin): + """Report every taint flow carrying ``kind`` into a sink.""" + + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({TaintAnalysis}) + rule_id: ClassVar[str] + kind: ClassVar[TaintKind] + severity: ClassVar[Severity] + title: ClassVar[str] + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + findings: list[Finding] = [] + for function in ctx.functions(): + for flow in ctx.get(TaintAnalysis, function).flows: + if not flow.kinds & self.kind: + continue + findings.append( + Finding( + rule_id=self.rule_id, + message=f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}", + severity=self.severity, + confidence=Confidence.HIGH, + span=flow.location, + function=function.name, + metadata={ + "source": str(flow.source.symbol), + "source_label": flow.source.label, + "sink": str(flow.sink.symbol), + }, + ) + ) + return findings + + +class SymbolCallDetector(Plugin): + """Report calls whose callee is one of ``symbols``, whatever name the file uses.""" + + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis}) + rule_id: ClassVar[str] + symbols: ClassVar[frozenset[SymbolId]] + severity: ClassVar[Severity] + message_template: ClassVar[str] + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + findings: list[Finding] = [] + for function in ctx.functions(): + ssa = ctx.get(SSAAnalysis, function) + defined = { + i.result: i.symbol_id + for block in ssa.blocks + for i in block.instructions + if isinstance(i, Symbol) + } + for block in ssa.blocks: + for instruction in block.instructions: + if not isinstance(instruction, Call): + continue + symbol = defined.get(instruction.callee) + if symbol is None or symbol not in self.symbols: + continue + findings.append( + Finding( + rule_id=self.rule_id, + message=self.message_template.format(symbol=symbol), + severity=self.severity, + confidence=Confidence.HIGH, + span=instruction.location, + function=function.name, + metadata={"symbol": str(symbol)}, + ) + ) + return findings diff --git a/tests/test_detectors.py b/tests/test_detectors.py new file mode 100644 index 0000000..2e8a0eb --- /dev/null +++ b/tests/test_detectors.py @@ -0,0 +1,302 @@ +"""Acceptance tests for the first detectors (``docs/architecture.md`` §15, §25, §35, §36). + +Detectors consume generic facts. Two reusable bases live in the plugin API: +``TaintDetector`` turns the flows of one taint kind into findings, and +``SymbolCallDetector`` reports calls resolving to a set of canonical symbols. Any plugin +may contribute security models through ``Plugin.models``; the engine collects them from +every loaded plugin before providing the table, so a model plugin and a detector plugin +compose without knowing each other. + +Shipped under ``plugins/``: a ``python-stdlib`` model plugin, the ``sql-injection``, +``command-injection``, ``path-traversal``, ``ssrf`` and ``xss`` taint detectors, and the +``dangerous-eval`` and ``weak-crypto`` syntax detectors. + +Expected to remain red until the detector bases, ``Plugin.models`` and the shipped +plugins exist. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import ClassVar + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.findings import Finding, Severity +from coretrace_python.plugins import Plugin, discover_plugins, run_plugins +from coretrace_python.reporters import render_sarif +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import SecurityModelRegistry, Sink, Source, TaintKind + +try: + from coretrace_python.plugins import ModelPlugin, SymbolCallDetector, TaintDetector +except ImportError as error: # pragma: no cover - red until detector bases land + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_detectors() -> None: + if MISSING is not None: + pytest.fail(f"detector bases are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def check(source_text: str, *extra_roots: Path, name: str = "app.py") -> tuple[Finding, ...]: + source = SourceManager().add_source(name, source_text) + return engine.check(source, [PLUGINS, *extra_roots]) + + +def rules(findings: tuple[Finding, ...]) -> list[str]: + return [f.rule_id for f in findings] + + +# --------------------------------------------------------------------------- bases + + +if MISSING is None: + + class CommandFlows(TaintDetector): + name: ClassVar[str] = "test.command" + rule_id: ClassVar[str] = "test-command" + kind: ClassVar[TaintKind] = TaintKind.COMMAND + severity: ClassVar[Severity] = Severity.CRITICAL + title: ClassVar[str] = "Command injection" + + class SqlFlows(TaintDetector): + name: ClassVar[str] = "test.sql" + rule_id: ClassVar[str] = "test-sql" + kind: ClassVar[TaintKind] = TaintKind.SQL + severity: ClassVar[Severity] = Severity.HIGH + title: ClassVar[str] = "SQL injection" + + class PrintCalls(SymbolCallDetector): + name: ClassVar[str] = "test.print" + rule_id: ClassVar[str] = "test-print" + symbols: ClassVar[frozenset[SymbolId]] = frozenset({SymbolId("python.builtins.print")}) + severity: ClassVar[Severity] = Severity.INFO + message_template: ClassVar[str] = "call to {symbol}" + + class WebModels(ModelPlugin): + name: ClassVar[str] = "test.web-models" + models = ( + Source(SymbolId("python.web.param"), "http"), + Sink(SymbolId("python.db.execute"), TaintKind.SQL), + Sink(SymbolId("python.web.render"), TaintKind.HTML), + ) + + +def manager_with_models(source_text: str): # type: ignore[no-untyped-def] + registry = SecurityModelRegistry() + registry.register(*WebModels.models, Sink(SymbolId("python.os.system"), TaintKind.COMMAND)) + module = engine.build_hir(SourceManager().add_source("app.py", source_text)) + return engine.build_manager(module, registry) + + +def test_taint_detector_reports_flows_of_its_kind_only() -> None: + manager = manager_with_models( + "import os\nimport db\nfrom web import param\n\n" + "def run():\n" + " value = param('x')\n" + " os.system(value)\n" + " db.execute(value)\n" + ) + + findings = run_plugins(manager, [CommandFlows(), SqlFlows()]) + + assert rules(findings) == ["test-command", "test-sql"] + command = findings[0] + assert command.severity is Severity.CRITICAL + assert command.span.start_line == 7 + assert command.function == "run" + assert command.message == "Command injection: http input reaches python.os.system" + assert dict(command.metadata) == { + "source": "python.web.param", + "source_label": "http", + "sink": "python.os.system", + } + + +def test_taint_detector_declares_only_the_taint_analysis() -> None: + from coretrace_python.taint import TaintAnalysis + + assert CommandFlows.requires == frozenset({TaintAnalysis}) + + +def test_symbol_call_detector_reports_resolved_calls() -> None: + manager = manager_with_models("def f(x):\n print(x)\n show = print\n show(x)\n") + + findings = run_plugins(manager, [PrintCalls()]) + + assert [(f.rule_id, f.span.start_line, f.message) for f in findings] == [ + ("test-print", 2, "call to python.builtins.print"), + ("test-print", 4, "call to python.builtins.print"), + ] + assert findings[0].severity is Severity.INFO + + +def test_plugins_may_carry_models_and_model_plugins_report_nothing() -> None: + assert Plugin.models == () + assert len(WebModels.models) == 3 + manager = manager_with_models("def f():\n pass\n") + assert run_plugins(manager, [WebModels()]) == () + + +# --------------------------------------------------------------------------- shipped plugins + + +def test_shipped_plugins_load_with_their_manifests() -> None: + module = engine.build_hir(SourceManager().add_source("empty.py", "")) + loaded = discover_plugins(PLUGINS, engine.build_manager(module)) + by_name = {plugin.manifest.name: plugin.manifest for plugin in loaded} + + assert set(by_name) == { + "command-injection", + "dangerous-eval", + "path-traversal", + "python-stdlib-models", + "sql-injection", + "ssrf", + "weak-crypto", + "xss", + } + for rule in ("sql-injection", "command-injection", "path-traversal", "ssrf", "xss"): + assert by_name[rule].requires == ("taint.flows",) + assert by_name[rule].provides == (f"vulnerability.{rule}",) + assert by_name["python-stdlib-models"].provides == ("model.python-stdlib",) + assert by_name["weak-crypto"].provides == ("vulnerability.weak-crypto",) + + +def test_stdin_to_os_system_is_a_command_injection() -> None: + findings = check("import os\n\ndef run():\n cmd = input()\n os.system(cmd)\n") + + assert rules(findings) == ["command-injection"] + assert findings[0].severity is Severity.HIGH + assert findings[0].span.start_line == 5 + assert findings[0].metadata["source_label"] == "stdin" + + +def test_shlex_quote_sanitizes_command_arguments() -> None: + findings = check( + "import os\nimport shlex\n\ndef run():\n cmd = shlex.quote(input())\n os.system(cmd)\n" + ) + assert findings == () + + +def test_argv_to_open_is_a_path_traversal() -> None: + findings = check("import sys\n\ndef read():\n return open(sys.argv[1])\n") + assert rules(findings) == ["path-traversal"] + + +def test_environment_to_urlopen_is_ssrf() -> None: + findings = check( + "import os\nfrom urllib.request import urlopen\n\n" + "def fetch():\n return urlopen(os.environ['TARGET'])\n" + ) + assert rules(findings) == ["ssrf"] + + +def test_one_flow_produces_one_finding() -> None: + findings = check("import subprocess\n\ndef run():\n subprocess.run(input())\n") + assert rules(findings) == ["command-injection"] + + +def test_weak_hashes_are_reported() -> None: + findings = check( + "import hashlib\n\ndef digest(data):\n a = hashlib.md5(data)\n b = hashlib.sha1(data)\n return hashlib.sha256(data)\n" + ) + + assert rules(findings) == ["weak-crypto", "weak-crypto"] + assert [f.span.start_line for f in findings] == [4, 5] + assert findings[0].severity is Severity.MEDIUM + assert "python.hashlib.md5" in findings[0].message + + +def test_dangerous_eval_still_reports_syntactically() -> None: + findings = check("def run(code):\n eval(code)\n") + assert rules(findings) == ["dangerous-eval"] + + +# --------------------------------------------------------------------------- composition + + +def write_model_plugin(directory: Path) -> Path: + plugin_dir = directory / "web_models" + plugin_dir.mkdir() + (plugin_dir / "plugin.toml").write_text( + 'name = "web-models"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\n' + 'requires = []\nprovides = ["model.http-sources", "model.sql-sinks", "model.html-sinks"]\n\n' + '[entrypoint]\nmodule = "web_models"\nclass = "WebModels"\n', + encoding="utf-8", + ) + (plugin_dir / "web_models.py").write_text( + "from typing import ClassVar\n\n" + "from coretrace_python.plugins import ModelPlugin\n" + "from coretrace_python.semantic.symbols import SymbolId\n" + "from coretrace_python.taint import Sink, Source, TaintKind\n\n\n" + "class WebModels(ModelPlugin):\n" + ' name: ClassVar[str] = "web-models"\n' + " models = (\n" + ' Source(SymbolId("python.web.param"), "http"),\n' + ' Sink(SymbolId("python.db.execute"), TaintKind.SQL),\n' + ' Sink(SymbolId("python.web.render"), TaintKind.HTML),\n' + " )\n", + encoding="utf-8", + ) + return directory + + +def test_model_plugins_compose_with_detectors(tmp_path: Path) -> None: + findings = check( + "import db\nfrom web import param, render\n\n" + "def lookup():\n" + " ident = param('id')\n" + " db.execute('SELECT * FROM users WHERE id = ' + ident)\n" + " render('' + ident + '')\n", + write_model_plugin(tmp_path), + ) + + assert rules(findings) == ["sql-injection", "xss"] + assert [f.span.start_line for f in findings] == [6, 7] + + +def test_without_the_model_plugin_the_same_code_is_silent() -> None: + findings = check( + "import db\nfrom web import param\n\n" + "def lookup():\n db.execute('SELECT ' + param('id'))\n" + ) + assert findings == () + + +def test_end_to_end_sarif_report(tmp_path: Path) -> None: + findings = check( + "import db\nfrom web import param\n\ndef lookup():\n db.execute(param('id'))\n", + write_model_plugin(tmp_path), + name="routes.py", + ) + + document = json.loads(render_sarif(engine.report(findings))) + (result,) = document["runs"][0]["results"] + + assert result["ruleId"] == "sql-injection" + assert result["level"] == "error" + assert result["locations"][0]["physicalLocation"]["artifactLocation"]["uri"] == "routes.py" + + +def test_cli_reports_taint_findings(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + source = tmp_path / "run.py" + source.write_text("import os\n\ndef run():\n os.system(input())\n", encoding="utf-8") + + exit_code = main(["--check", str(source), "--plugins", str(PLUGINS)]) + output = capsys.readouterr().out + + assert exit_code == 1 + assert output.startswith(f"{source}:4:5: high command-injection: Command injection: stdin input reaches python.os.system [run]\n") diff --git a/tests/test_plugins.py b/tests/test_plugins.py index 27b6ea7..3bc564a 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -67,9 +67,13 @@ def require_plugin_api() -> None: def manager_for(source_text: str) -> AnalysisManager: + from coretrace_python.cfg import CFGAnalysis, DominanceAnalysis + from coretrace_python.ir.lowering import PyIRAnalysis + from coretrace_python.ir.ssa import SSAAnalysis + module = build_hir(SourceManager().add_source("plugged.py", source_text)) manager = AnalysisManager(module) - manager.register(*SEMANTIC_ANALYSES) + manager.register(*SEMANTIC_ANALYSES, CFGAnalysis, DominanceAnalysis, PyIRAnalysis, SSAAnalysis) return manager @@ -224,7 +228,7 @@ def test_reference_manifest_is_parsed() -> None: 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.requires == ("ir.ssa",) assert manifest.provides == ("vulnerability.dangerous-eval",) assert manifest.entrypoint.module == "dangerous_eval" assert manifest.entrypoint.class_name == "DangerousEvalPlugin" @@ -297,7 +301,9 @@ def test_loads_the_reference_plugin() -> None: assert loaded.manifest.name == "dangerous-eval" assert isinstance(loaded.plugin, Plugin) assert loaded.plugin.name == "dangerous-eval" - assert loaded.plugin.requires == frozenset({ScopeAnalysis, SymbolAnalysis}) + from coretrace_python.ir.ssa import SSAAnalysis + + assert loaded.plugin.requires == frozenset({SSAAnalysis}) def test_incompatible_plugin_api_is_rejected(tmp_path: Path) -> None: