From edb43c44dbc197540df06b58253bdbe5aaf3d6f0 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 02:06:28 +0200 Subject: [PATCH 1/2] test: define dependency analysis behavior --- tests/test_architecture.py | 5 +- tests/test_dependencies.py | 345 +++++++++++++++++++++++++++++++++++++ tests/test_detectors.py | 3 + 3 files changed, 351 insertions(+), 2 deletions(-) create mode 100644 tests/test_dependencies.py diff --git a/tests/test_architecture.py b/tests/test_architecture.py index e779b04..ff94c27 100644 --- a/tests/test_architecture.py +++ b/tests/test_architecture.py @@ -36,8 +36,9 @@ "interprocedural": 7, "taint": 8, "findings": 9, - "plugins": 10, - "reporters": 11, + "dependency": 10, + "plugins": 11, + "reporters": 12, } # Top-level modules that wire the pipeline together and may import any layer. diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py new file mode 100644 index 0000000..4153419 --- /dev/null +++ b/tests/test_dependencies.py @@ -0,0 +1,345 @@ +"""Acceptance tests for dependency analysis (``docs/architecture.md`` §26). + +The engine resolves the requirements declared in ``requirements.txt``, +``pyproject.toml``, ``poetry.lock`` and ``uv.lock`` into a ``DependencyGraph`` and +provides it to the analyses as the ``dependency.graph`` input. Plugins may contribute +``Advisory`` records (package, vulnerable range, affected APIs) and may run once per +project as ``ProjectPlugin``s with a ``ProjectContext`` that exposes the module graph, +the dependency graph, the advisories and every module's imports. + +Shipped under ``plugins/dependency``: a sample advisory database, the +``vulnerable-dependency`` detector (a requirement allows a vulnerable version) and the +``reachable-vulnerability`` detector (an affected API is imported by a project module). + +Expected to remain red until ``coretrace_python.dependency`` and the project plugin +kind exist. +""" + +from __future__ import annotations + +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 Confidence, Finding, Severity +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceId, SourceManager + +try: + from coretrace_python.dependency import ( + Advisory, + DependencyAnalysis, + DependencyGraph, + Requirement, + Version, + parse_dependencies, + ) + from coretrace_python.plugins import ProjectContext, ProjectPlugin +except ImportError as error: # pragma: no cover - red until dependency analysis lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_dependencies() -> None: + if MISSING is not None: + pytest.fail(f"dependency analysis is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def project(root: Path, files: dict[str, str]) -> Path: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +def source(name: str, text: str): # type: ignore[no-untyped-def] + return SourceManager().add_source(name, text) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: + return [(Path(str(f.span.source_id)).name, f.rule_id, f.span.start_line) for f in findings] + + +# --------------------------------------------------------------------------- parsing + + +def test_requirements_txt_lines_become_requirements() -> None: + graph = parse_dependencies( + source( + "requirements.txt", + "# pinned\nPyYAML==5.3.1\nrequests[security]>=2.18, <3\n-r other.txt\n" + "flask ; python_version >= '3.8'\n\nDjango~=4.2.0 # lts\n", + ) + ) + + assert isinstance(graph, DependencyGraph) + assert graph.names == ("django", "flask", "pyyaml", "requests") + yaml = graph.requirement("PyYAML") + assert isinstance(yaml, Requirement) + assert yaml.name == "pyyaml" + assert yaml.specifier == "==5.3.1" + assert yaml.pinned == Version.parse("5.3.1") + assert yaml.span.source_id == SourceId("requirements.txt") + assert yaml.span.start_line == 2 + assert graph.requirement("requests").specifier == ">=2.18,<3" # type: ignore[union-attr] + assert graph.requirement("requests").pinned is None # type: ignore[union-attr] + assert graph.requirement("flask").specifier == "" # type: ignore[union-attr] + assert graph.requirement("django").specifier == "~=4.2.0" # type: ignore[union-attr] + + +def test_pyproject_dependencies_pep621_and_poetry() -> None: + graph = parse_dependencies( + source( + "pyproject.toml", + '[project]\nname = "app"\ndependencies = ["requests>=2.20", "pyyaml"]\n\n' + '[project.optional-dependencies]\ndev = ["pytest>=8"]\n\n' + '[tool.poetry.dependencies]\npython = "^3.11"\njinja2 = "^2.11"\nflask = { version = ">=2.0", extras = ["async"] }\n', + ) + ) + + assert graph.names == ("flask", "jinja2", "pytest", "pyyaml", "requests") + assert graph.requirement("jinja2").specifier == "^2.11" # type: ignore[union-attr] + assert graph.requirement("flask").specifier == ">=2.0" # type: ignore[union-attr] + assert graph.requirement("pytest").optional is True # type: ignore[union-attr] + assert graph.requirement("requests").optional is False # type: ignore[union-attr] + assert graph.requirement("requests").span.start_line == 3 # type: ignore[union-attr] + + +def test_lockfiles_pin_versions() -> None: + poetry = parse_dependencies( + source("poetry.lock", '[[package]]\nname = "pyyaml"\nversion = "5.3.1"\n\n[[package]]\nname = "Requests"\nversion = "2.31.0"\n') + ) + uv = parse_dependencies( + source("uv.lock", 'version = 1\n\n[[package]]\nname = "urllib3"\nversion = "1.26.4"\nsource = { registry = "https://pypi.org/simple" }\n') + ) + + assert poetry.requirement("pyyaml").pinned == Version.parse("5.3.1") # type: ignore[union-attr] + assert poetry.requirement("requests").pinned == Version.parse("2.31.0") # type: ignore[union-attr] + assert uv.requirement("urllib3").pinned == Version.parse("1.26.4") # type: ignore[union-attr] + assert uv.requirement("urllib3").span.start_line == 4 # type: ignore[union-attr] + + +def test_unknown_files_and_malformed_content_are_ignored_or_reported() -> None: + assert parse_dependencies(source("setup.py", "x = 1\n")).names == () + graph = parse_dependencies(source("pyproject.toml", "not = [toml\n")) + assert graph.names == () + assert graph.errors and "pyproject.toml" in graph.errors[0] + + +def test_graphs_merge_and_locks_pin_declared_requirements() -> None: + declared = parse_dependencies(source("requirements.txt", "pyyaml>=5\n")) + locked = parse_dependencies(source("poetry.lock", '[[package]]\nname = "pyyaml"\nversion = "5.3.1"\n')) + + merged = declared.merge(locked) + + assert merged.requirement("pyyaml").specifier == ">=5" # type: ignore[union-attr] + assert merged.requirement("pyyaml").pinned == Version.parse("5.3.1") # type: ignore[union-attr] + assert merged.names == ("pyyaml",) + + +# --------------------------------------------------------------------------- versions + + +@pytest.mark.parametrize( + "version, specifier, expected", + [ + ("5.3.1", "<5.4", True), + ("5.4", "<5.4", False), + ("2.19.1", ">=2.18,<2.20.0", True), + ("2.20.0", ">=2.18,<2.20.0", False), + ("4.2.7", "~=4.2.0", True), + ("4.3.0", "~=4.2.0", False), + ("2.11.3", "==2.11.3", True), + ("1.26.4", ">=1.26.5", False), + ("2.11.0", "^2.11", True), + ("3.0.0", "^2.11", False), + ], +) +def test_version_matching(version: str, specifier: str, expected: bool) -> None: + assert Version.parse(version).satisfies(specifier) is expected + + +def test_requirements_may_allow_a_vulnerable_version() -> None: + pinned = Requirement.parse("pyyaml==5.3.1", SourceId("r.txt"), 1) + floor = Requirement.parse("pyyaml>=5.1", SourceId("r.txt"), 1) + safe = Requirement.parse("pyyaml>=5.4", SourceId("r.txt"), 1) + unspecified = Requirement.parse("pyyaml", SourceId("r.txt"), 1) + + assert pinned.may_match("<5.4") is True + assert Requirement.parse("pyyaml==6.0", SourceId("r.txt"), 1).may_match("<5.4") is False + assert floor.may_match("<5.4") is True + assert safe.may_match("<5.4") is False + assert unspecified.may_match("<5.4") is True + + +# --------------------------------------------------------------------------- engine + + +def test_projects_resolve_their_dependency_files(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "pyyaml>=5\nrequests==2.31.0\n", + "poetry.lock": '[[package]]\nname = "pyyaml"\nversion = "5.3.1"\n', + "app/__init__.py": "", + "app/main.py": "import yaml\n", + }, + ) + + result = engine.analyze_project(root) + + assert result.dependencies.names == ("pyyaml", "requests") + assert result.dependencies.requirement("pyyaml").pinned == Version.parse("5.3.1") # type: ignore[union-attr] + assert DependencyAnalysis.name == "dependency.graph" + + +def test_single_files_see_an_empty_dependency_graph() -> None: + manager = engine.build_manager(engine.build_hir(source("x.py", ""))) + assert manager.get(DependencyAnalysis).names == () + + +if MISSING is None: + + class CountRequirements(ProjectPlugin): + name: ClassVar[str] = "test.count-requirements" + runs: ClassVar[int] = 0 + + def analyze_project(self, ctx: ProjectContext) -> list[Finding]: + CountRequirements.runs += 1 + return [ + Finding( + "test.requirement", + f"{requirement.name} in {len(ctx.graph.modules)} modules", + Severity.INFO, + Confidence.HIGH, + requirement.span, + ) + for requirement in ctx.dependencies.requirements + ] + + +def test_project_plugins_run_once_per_project(tmp_path: Path) -> None: + root = project( + tmp_path, + {"requirements.txt": "pyyaml\nrequests\n", "a.py": "", "b.py": ""}, + ) + CountRequirements.runs = 0 + + findings = engine.analyze_project(root, plugins=[CountRequirements()]).findings + + assert CountRequirements.runs == 1 + assert [(f.rule_id, f.message) for f in findings] == [ + ("test.requirement", "pyyaml in 2 modules"), + ("test.requirement", "requests in 2 modules"), + ] + assert rules(findings) == [("requirements.txt", "test.requirement", 1), ("requirements.txt", "test.requirement", 2)] + + +# --------------------------------------------------------------------------- shipped plugins + + +def test_advisory_records() -> None: + advisory = Advisory( + id="CVE-2020-1747", + package="pyyaml", + vulnerable="<5.4", + summary="yaml.load can execute arbitrary code", + severity=Severity.CRITICAL, + affected_symbols=(SymbolId("python.yaml.load"), SymbolId("python.yaml.full_load")), + ) + assert advisory.affects(Requirement.parse("pyyaml==5.3.1", SourceId("r"), 1)) + assert not advisory.affects(Requirement.parse("pyyaml==6.0", SourceId("r"), 1)) + + +def test_sample_advisories_are_shipped() -> None: + from coretrace_python.plugins import discover_plugins + + loaded = {p.manifest.name: p for p in discover_plugins(PLUGINS, engine.build_manager(engine.build_hir(source("e.py", ""))))} + + assert loaded["sample-advisories"].manifest.provides == ("advisories.sample",) + advisories = loaded["sample-advisories"].plugin.advisories + assert any(a.package == "pyyaml" and "CVE" in a.id for a in advisories) + assert loaded["vulnerable-dependency"].manifest.requires == ("dependency.graph",) + assert loaded["reachable-vulnerability"].manifest.requires == ("dependency.graph", "interprocedural.callgraph") + + +def test_vulnerable_requirement_is_reported_at_its_line(tmp_path: Path) -> None: + root = project( + tmp_path, + {"requirements.txt": "requests==2.31.0\npyyaml==5.3.1\n", "app.py": "x = 1\n"}, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("requirements.txt", "vulnerable-dependency", 2)] + assert findings[0].severity is Severity.CRITICAL + assert findings[0].confidence is Confidence.HIGH + assert "CVE-2020-1747" in findings[0].message + assert findings[0].metadata["advisory"] == "CVE-2020-1747" + assert findings[0].metadata["package"] == "pyyaml" + + +def test_unpinned_requirements_that_allow_a_vulnerable_version_are_medium_confidence(tmp_path: Path) -> None: + root = project(tmp_path, {"requirements.txt": "pyyaml>=5.1\n", "app.py": ""}) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("requirements.txt", "vulnerable-dependency", 1)] + assert findings[0].confidence is Confidence.MEDIUM + + +def test_safe_versions_are_silent(tmp_path: Path) -> None: + root = project(tmp_path, {"requirements.txt": "pyyaml>=6.0\nrequests>=2.31\n", "app.py": ""}) + assert engine.analyze_project(root, [PLUGINS]).findings == () + + +def test_imported_affected_api_is_a_reachable_vulnerability(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "pyyaml==5.3.1\n", + "app/__init__.py": "", + "app/config.py": "import yaml\n\ndef load(text):\n return yaml.load(text)\n", + "app/other.py": "from yaml import safe_load\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [ + ("config.py", "reachable-vulnerability", 4), + ("requirements.txt", "vulnerable-dependency", 1), + ] + reachable = findings[0] + assert reachable.severity is Severity.CRITICAL + assert reachable.function == "load" + assert reachable.metadata["symbol"] == "python.yaml.load" + assert reachable.metadata["advisory"] == "CVE-2020-1747" + + +def test_reachability_needs_a_vulnerable_requirement(tmp_path: Path) -> None: + root = project( + tmp_path, + {"requirements.txt": "pyyaml==6.0\n", "app.py": "import yaml\n\ndef load(t):\n return yaml.load(t)\n"}, + ) + assert engine.analyze_project(root, [PLUGINS]).findings == () + + +def test_cli_reports_dependency_findings(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = project(tmp_path, {"requirements.txt": "pyyaml==5.3.1\n", "app.py": ""}) + + exit_code = main(["--check", str(root), "--plugins", str(PLUGINS)]) + output = capsys.readouterr().out + + assert exit_code == 1 + assert output.startswith(f"{(root / 'requirements.txt').resolve()}:1:1: critical vulnerable-dependency:") diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 3f9fb17..9995f1f 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -168,9 +168,12 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "flask-models", "path-traversal", "python-stdlib-models", + "reachable-vulnerability", + "sample-advisories", "sql-injection", "sqlalchemy-models", "ssrf", + "vulnerable-dependency", "weak-crypto", "xss", } From cf01d4907486436c31f75ccadb97315360315a77 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 02:06:28 +0200 Subject: [PATCH 2/2] feat: add dependency resolution, advisories and project plugins --- README.md | 5 + .../reachable_vulnerability/plugin.toml | 9 + .../reachable_vulnerability.py | 56 ++++ .../dependency/sample_advisories/plugin.toml | 9 + .../sample_advisories/sample_advisories.py | 64 +++++ .../vulnerable_dependency/plugin.toml | 9 + .../vulnerable_dependency.py | 43 +++ src/coretrace_python/dependency/__init__.py | 23 ++ src/coretrace_python/dependency/graph.py | 272 ++++++++++++++++++ src/coretrace_python/engine.py | 72 +++-- src/coretrace_python/plugins/__init__.py | 4 + src/coretrace_python/plugins/api.py | 46 ++- 12 files changed, 594 insertions(+), 18 deletions(-) create mode 100644 plugins/dependency/reachable_vulnerability/plugin.toml create mode 100644 plugins/dependency/reachable_vulnerability/reachable_vulnerability.py create mode 100644 plugins/dependency/sample_advisories/plugin.toml create mode 100644 plugins/dependency/sample_advisories/sample_advisories.py create mode 100644 plugins/dependency/vulnerable_dependency/plugin.toml create mode 100644 plugins/dependency/vulnerable_dependency/vulnerable_dependency.py create mode 100644 src/coretrace_python/dependency/__init__.py create mode 100644 src/coretrace_python/dependency/graph.py diff --git a/README.md b/README.md index e22b9fd..2993f9f 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,11 @@ Given a directory, every Python file below it is analysed as one project: module named after their packages, and taint follows calls into functions defined in other files through a project-wide index of function summaries iterated to a fixpoint. Hidden and tooling directories (`.venv`, `node_modules`, `__pycache__`, `build`, `dist`) are skipped. +The dependency files at the root (`requirements*.txt`, `pyproject.toml`, `poetry.lock`, +`uv.lock`) are resolved into a dependency graph. Plugins may contribute advisories; the +shipped `dependency/` plugins report a requirement that allows a vulnerable version at +its line, and every call in the project to an API the advisory affects. The shipped +advisory database is a small offline sample; a live OSV feed is future work. `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be repeated. The repository ships a first set under `plugins/`: security models for the diff --git a/plugins/dependency/reachable_vulnerability/plugin.toml b/plugins/dependency/reachable_vulnerability/plugin.toml new file mode 100644 index 0000000..22d99f9 --- /dev/null +++ b/plugins/dependency/reachable_vulnerability/plugin.toml @@ -0,0 +1,9 @@ +name = "reachable-vulnerability" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["dependency.graph", "interprocedural.callgraph"] +provides = ["vulnerability.reachable-vulnerability"] + +[entrypoint] +module = "reachable_vulnerability" +class = "ReachableVulnerabilityPlugin" diff --git a/plugins/dependency/reachable_vulnerability/reachable_vulnerability.py b/plugins/dependency/reachable_vulnerability/reachable_vulnerability.py new file mode 100644 index 0000000..a76f5d9 --- /dev/null +++ b/plugins/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -0,0 +1,56 @@ +"""Calls, anywhere in the project, to an API affected by a vulnerable requirement.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.dependency import DependencyAnalysis +from coretrace_python.findings import Confidence, Finding +from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol +from coretrace_python.plugins import ProjectContext, ProjectPlugin + + +class ReachableVulnerabilityPlugin(ProjectPlugin): + name: ClassVar[str] = "reachable-vulnerability" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({DependencyAnalysis, CallGraphAnalysis}) + + def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: + affected = {} + for requirement in ctx.dependencies.requirements: + for advisory in ctx.advisories: + if advisory.affects(requirement): + for symbol in advisory.affected_symbols: + affected.setdefault(symbol, advisory) + if not affected: + return () + findings: list[Finding] = [] + for module in ctx.modules: + graph = ctx.call_graph(module) + for function in graph.functions: + for site in graph.sites(function): + if not isinstance(site.target, ExternalSymbol): + continue + advisory = affected.get(site.target.symbol) + if advisory is None: + continue + findings.append( + Finding( + rule_id="reachable-vulnerability", + message=( + f"{advisory.id}: {site.target.symbol} is affected in the required " + f"{advisory.package} {advisory.vulnerable}: {advisory.summary}" + ), + severity=advisory.severity, + confidence=Confidence.HIGH, + span=site.location, + function=function, + metadata={ + "advisory": advisory.id, + "package": advisory.package, + "symbol": str(site.target.symbol), + }, + ) + ) + return findings diff --git a/plugins/dependency/sample_advisories/plugin.toml b/plugins/dependency/sample_advisories/plugin.toml new file mode 100644 index 0000000..746efbb --- /dev/null +++ b/plugins/dependency/sample_advisories/plugin.toml @@ -0,0 +1,9 @@ +name = "sample-advisories" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["advisories.sample"] + +[entrypoint] +module = "sample_advisories" +class = "SampleAdvisories" diff --git a/plugins/dependency/sample_advisories/sample_advisories.py b/plugins/dependency/sample_advisories/sample_advisories.py new file mode 100644 index 0000000..cd9757f --- /dev/null +++ b/plugins/dependency/sample_advisories/sample_advisories.py @@ -0,0 +1,64 @@ +"""A small, offline sample of published advisories (architecture §26). + +This is a demonstration database: a handful of well-known CVEs with the APIs they +affect. A live OSV or GHSA feed belongs to a dedicated plugin. +""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.dependency import Advisory +from coretrace_python.findings import Severity +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class SampleAdvisories(ModelPlugin): + name: ClassVar[str] = "sample-advisories" + advisories: ClassVar[tuple[Advisory, ...]] = ( + Advisory( + "CVE-2020-1747", + "pyyaml", + "<5.4", + "yaml.load and full_load can execute arbitrary code from untrusted documents", + Severity.CRITICAL, + (_sym("yaml.load"), _sym("yaml.full_load"), _sym("yaml.unsafe_load")), + ), + Advisory( + "CVE-2018-18074", + "requests", + "<2.20.0", + "Authorization header is leaked on redirects to another host", + Severity.HIGH, + (_sym("requests.get"), _sym("requests.post"), _sym("requests.request"), _sym("requests.Session")), + ), + Advisory( + "CVE-2020-28493", + "jinja2", + "<2.11.3", + "regular expression denial of service in the urlize filter", + Severity.MEDIUM, + (_sym("jinja2.Environment"), _sym("jinja2.Template")), + ), + Advisory( + "CVE-2021-33503", + "urllib3", + "<1.26.5", + "regular expression denial of service when parsing authority in URLs", + Severity.HIGH, + (_sym("urllib3.PoolManager"), _sym("urllib3.request")), + ), + Advisory( + "CVE-2023-30861", + "flask", + "<2.2.5", + "session cookie may be leaked through caching proxies", + Severity.HIGH, + (_sym("flask.Flask"),), + ), + ) diff --git a/plugins/dependency/vulnerable_dependency/plugin.toml b/plugins/dependency/vulnerable_dependency/plugin.toml new file mode 100644 index 0000000..1e94a53 --- /dev/null +++ b/plugins/dependency/vulnerable_dependency/plugin.toml @@ -0,0 +1,9 @@ +name = "vulnerable-dependency" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["dependency.graph"] +provides = ["vulnerability.vulnerable-dependency"] + +[entrypoint] +module = "vulnerable_dependency" +class = "VulnerableDependencyPlugin" diff --git a/plugins/dependency/vulnerable_dependency/vulnerable_dependency.py b/plugins/dependency/vulnerable_dependency/vulnerable_dependency.py new file mode 100644 index 0000000..778645d --- /dev/null +++ b/plugins/dependency/vulnerable_dependency/vulnerable_dependency.py @@ -0,0 +1,43 @@ +"""Requirements that allow a version an advisory marks as vulnerable.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.dependency import DependencyAnalysis +from coretrace_python.findings import Confidence, Finding +from coretrace_python.plugins import ProjectContext, ProjectPlugin + + +class VulnerableDependencyPlugin(ProjectPlugin): + name: ClassVar[str] = "vulnerable-dependency" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({DependencyAnalysis}) + + def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: + findings: list[Finding] = [] + for requirement in ctx.dependencies.requirements: + for advisory in ctx.advisories: + if not advisory.affects(requirement): + continue + pinned = requirement.pinned is not None + findings.append( + Finding( + rule_id="vulnerable-dependency", + message=( + f"{advisory.id}: {advisory.package} {advisory.vulnerable} is " + f"{'required' if pinned else 'allowed'} by {requirement.name}" + f"{requirement.specifier or ''}: {advisory.summary}" + ), + severity=advisory.severity, + confidence=Confidence.HIGH if pinned else Confidence.MEDIUM, + span=requirement.span, + metadata={ + "advisory": advisory.id, + "package": advisory.package, + "specifier": requirement.specifier, + }, + ) + ) + return findings diff --git a/src/coretrace_python/dependency/__init__.py b/src/coretrace_python/dependency/__init__.py new file mode 100644 index 0000000..28308b8 --- /dev/null +++ b/src/coretrace_python/dependency/__init__.py @@ -0,0 +1,23 @@ +"""Dependency resolution and advisories (architecture §26).""" + +from coretrace_python.dependency.graph import ( + DEPENDENCY_FILES, + Advisory, + DependencyAnalysis, + DependencyGraph, + Requirement, + Version, + normalize, + parse_dependencies, +) + +__all__ = [ + "DEPENDENCY_FILES", + "Advisory", + "DependencyAnalysis", + "DependencyGraph", + "Requirement", + "Version", + "normalize", + "parse_dependencies", +] diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py new file mode 100644 index 0000000..a38bae6 --- /dev/null +++ b/src/coretrace_python/dependency/graph.py @@ -0,0 +1,272 @@ +"""Dependency resolution (architecture §26). + +Requirements declared in ``requirements.txt``, ``pyproject.toml`` (PEP 621 and Poetry) +and pinned in ``poetry.lock`` or ``uv.lock`` become a ``DependencyGraph``. Versions are +compared with a small PEP 440 subset (``==``, ``!=``, ``<``, ``<=``, ``>``, ``>=``, +``~=`` and Poetry's ``^``), enough to decide whether a requirement may allow a version an +advisory marks as vulnerable. Nothing is downloaded. +""" + +from __future__ import annotations + +import re +import tomllib +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import Any, ClassVar + +from coretrace_python.analysis import Analysis, AnalysisContext +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceFile, SourceId, SourceSpan + +_REQUIREMENT = re.compile(r"^\s*([A-Za-z0-9][A-Za-z0-9._-]*)\s*(\[[^\]]*\])?\s*(.*)$") +_CLAUSE = re.compile(r"^(===|==|!=|<=|>=|~=|<|>|\^)\s*([0-9][0-9A-Za-z.*+!-]*)$") + + +def normalize(name: str) -> str: + return re.sub(r"[-_.]+", "-", name).lower() + + +@dataclass(frozen=True, order=True) +class Version: + parts: tuple[int, ...] + + @classmethod + def parse(cls, text: str) -> Version: + parts: list[int] = [] + for piece in text.strip().split("."): + digits = re.match(r"\d+", piece) + if digits is None: + break + parts.append(int(digits.group())) + if digits.end() != len(piece): + break + return cls(tuple(parts) or (0,)) + + def padded(self, length: int) -> tuple[int, ...]: + return self.parts + (0,) * (length - len(self.parts)) + + def satisfies(self, specifier: str) -> bool: + for clause in [c.strip() for c in specifier.split(",") if c.strip()]: + match = _CLAUSE.match(clause) + if match is None: + continue + operator, wanted = match.group(1), Version.parse(match.group(2)) + if not self._satisfies_clause(operator, wanted, match.group(2)): + return False + return True + + def _satisfies_clause(self, operator: str, wanted: Version, raw: str) -> bool: + length = max(len(self.parts), len(wanted.parts)) + mine, theirs = self.padded(length), wanted.padded(length) + if operator in ("==", "==="): + if raw.endswith(".*"): + prefix = wanted.parts + return self.parts[: len(prefix)] == prefix + return mine == theirs + if operator == "!=": + return mine != theirs + if operator == "<": + return mine < theirs + if operator == "<=": + return mine <= theirs + if operator == ">": + return mine > theirs + if operator == ">=": + return mine >= theirs + if operator == "~=": + ceiling = list(wanted.parts[:-1]) if len(wanted.parts) > 1 else [wanted.parts[0]] + ceiling[-1] += 1 + return mine >= theirs and self.padded(len(ceiling)) < tuple(ceiling) + if operator == "^": + ceiling = [0] * len(wanted.parts) + for index, part in enumerate(wanted.parts): + if part != 0 or index == len(wanted.parts) - 1: + ceiling[index] = part + 1 + break + return mine >= theirs and self.padded(len(ceiling)) < tuple(ceiling) + return True + + +def _lower_bounds(specifier: str) -> list[Version]: + bounds: list[Version] = [] + for clause in [c.strip() for c in specifier.split(",") if c.strip()]: + match = _CLAUSE.match(clause) + if match is not None and match.group(1) in (">=", ">", "~=", "^", "==", "==="): + bounds.append(Version.parse(match.group(2))) + return bounds + + +@dataclass(frozen=True) +class Requirement: + name: str + specifier: str + span: SourceSpan + pinned: Version | None = None + optional: bool = False + + @classmethod + def parse(cls, text: str, source_id: SourceId, line: int, optional: bool = False) -> Requirement | None: + cleaned = text.split("#", 1)[0].split(";", 1)[0].strip() + match = _REQUIREMENT.match(cleaned) + if match is None or not match.group(1): + return None + specifier = ",".join(part.strip() for part in match.group(3).split(",") if part.strip()) + pinned = None + clauses = [c for c in specifier.split(",") if c] + if len(clauses) == 1 and clauses[0].startswith("==") and not clauses[0].endswith(".*"): + pinned = Version.parse(clauses[0].lstrip("=")) + return cls(normalize(match.group(1)), specifier, SourceSpan(source_id, line, 1), pinned, optional) + + def may_match(self, vulnerable: str) -> bool: + """Whether some version this requirement allows is in the ``vulnerable`` range.""" + + if self.pinned is not None: + return self.pinned.satisfies(vulnerable) + candidates = [Version((0,)), *_lower_bounds(self.specifier), *_lower_bounds(vulnerable)] + return any(v.satisfies(self.specifier) and v.satisfies(vulnerable) for v in candidates) + + +@dataclass(frozen=True) +class Advisory: + id: str + package: str + vulnerable: str + summary: str + severity: Severity + affected_symbols: tuple[SymbolId, ...] = () + + def affects(self, requirement: Requirement) -> bool: + return requirement.name == normalize(self.package) and requirement.may_match(self.vulnerable) + + +class DependencyGraph: + def __init__(self, requirements: Mapping[str, Requirement] | None = None, errors: tuple[str, ...] = ()) -> None: + self._requirements = MappingProxyType(dict(sorted((requirements or {}).items()))) + self.errors = errors + + @property + def names(self) -> tuple[str, ...]: + return tuple(self._requirements) + + @property + def requirements(self) -> tuple[Requirement, ...]: + return tuple(self._requirements.values()) + + def requirement(self, name: str) -> Requirement | None: + return self._requirements.get(normalize(name)) + + def merge(self, other: DependencyGraph) -> DependencyGraph: + merged = dict(self._requirements) + for name, requirement in other._requirements.items(): + current = merged.get(name) + if current is None: + merged[name] = requirement + continue + merged[name] = Requirement( + name, + current.specifier or requirement.specifier, + current.span if current.specifier else requirement.span, + requirement.pinned or current.pinned, + current.optional and requirement.optional, + ) + return DependencyGraph(merged, self.errors + other.errors) + + +def parse_dependencies(source: SourceFile) -> DependencyGraph: + """The requirements declared or pinned by one dependency file; other files are empty.""" + + name = source.path.name if source.path is not None else str(source.source_id) + if name.startswith("requirements") and name.endswith(".txt"): + return _parse_requirements_txt(source) + if name == "pyproject.toml": + return _parse_toml(source, _pyproject_requirements) + if name in ("poetry.lock", "uv.lock"): + return _parse_toml(source, _lock_requirements) + return DependencyGraph() + + +def _parse_requirements_txt(source: SourceFile) -> DependencyGraph: + found: dict[str, Requirement] = {} + for number, line in enumerate(source.text.splitlines(), start=1): + stripped = line.strip() + if not stripped or stripped.startswith(("#", "-")): + continue + requirement = Requirement.parse(stripped, source.source_id, number) + if requirement is not None: + found[requirement.name] = requirement + return DependencyGraph(found) + + +def _parse_toml(source: SourceFile, extract: Any) -> DependencyGraph: + try: + data = tomllib.loads(source.text) + except tomllib.TOMLDecodeError as error: + return DependencyGraph(errors=(f"{source.source_id}: {error}",)) + found: dict[str, Requirement] = {} + for requirement in extract(data, source): + found[requirement.name] = requirement + return DependencyGraph(found) + + +def _line_of(source: SourceFile, key: str, default: int = 1) -> int: + for number, line in enumerate(source.text.splitlines(), start=1): + if key in line: + return number + return default + + +def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]: + found: list[Requirement] = [] + project = data.get("project", {}) + for text in project.get("dependencies", []) or []: + requirement = Requirement.parse(text, source.source_id, _line_of(source, text)) + if requirement is not None: + found.append(requirement) + for group in (project.get("optional-dependencies", {}) or {}).values(): + for text in group or []: + requirement = Requirement.parse(text, source.source_id, _line_of(source, text), True) + if requirement is not None: + found.append(requirement) + poetry = data.get("tool", {}).get("poetry", {}) + for section, optional in (("dependencies", False), ("dev-dependencies", True)): + for name, spec in (poetry.get(section, {}) or {}).items(): + if normalize(name) == "python": + continue + specifier = spec.get("version", "") if isinstance(spec, dict) else str(spec) + if specifier == "*": + specifier = "" + requirement = Requirement.parse( + f"{name}{specifier}", source.source_id, _line_of(source, name), optional + ) + if requirement is not None: + found.append(requirement) + return found + + +def _lock_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]: + found: list[Requirement] = [] + for package in data.get("package", []) or []: + name, version = package.get("name"), package.get("version") + if not isinstance(name, str) or not isinstance(version, str): + continue + line = _line_of(source, f'name = "{name}"') + found.append( + Requirement(normalize(name), "", SourceSpan(source.source_id, line, 1), Version.parse(version)) + ) + return found + + +class DependencyAnalysis(Analysis[DependencyGraph]): + """The project's dependency graph, provided by the engine; empty for a lone file.""" + + name: ClassVar[str] = "dependency.graph" + + @classmethod + def compute(cls, ctx: AnalysisContext) -> DependencyGraph: + return DependencyGraph() + + +DEPENDENCY_FILES = ("pyproject.toml", "poetry.lock", "uv.lock") diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 45d5b3f..7bae9b9 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -6,8 +6,8 @@ from __future__ import annotations -from collections.abc import Sequence -from dataclasses import dataclass +from collections.abc import Iterable, Sequence +from dataclasses import dataclass, field from pathlib import Path from typing import ClassVar @@ -20,6 +20,12 @@ TransformationPass, ) from coretrace_python.cfg import CFGAnalysis, CFGError, DominanceAnalysis, PostDominanceAnalysis +from coretrace_python.dependency import ( + DEPENDENCY_FILES, + DependencyAnalysis, + DependencyGraph, + parse_dependencies, +) from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.findings.refutation import RefutationAnalysis from coretrace_python.frontend import HIRBuildError, ParseError, build_hir @@ -43,7 +49,14 @@ analyzable_functions, ) from coretrace_python.ir.ssa import SSAAnalysis -from coretrace_python.plugins import LoadedPlugin, PluginRegistry, discover_plugins, run_plugins +from coretrace_python.plugins import ( + Plugin, + PluginRegistry, + ProjectContext, + ProjectPlugin, + discover_plugins, + run_plugins, +) from coretrace_python.reporters import Report from coretrace_python.semantic import SEMANTIC_ANALYSES from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable @@ -74,6 +87,7 @@ SecurityModelAnalysis, TaintAnalysis, RefutationAnalysis, + DependencyAnalysis, ) @@ -105,6 +119,7 @@ class ProjectAnalysis: graph: ModuleGraph index: SummaryIndex findings: tuple[Finding, ...] + dependencies: DependencyGraph = field(default_factory=DependencyGraph) def build_manager( @@ -126,10 +141,10 @@ def load_plugins(plugin_roots: Sequence[Path], manager: AnalysisManager) -> Plug return registry -def plugin_models(registry: PluginRegistry) -> ModelTable: +def plugin_models(plugins: Iterable[Plugin]) -> ModelTable: models = SecurityModelRegistry() - for loaded in registry: - models.register(*loaded.plugin.models) + for plugin in plugins: + models.register(*plugin.models) return models.freeze() @@ -138,16 +153,34 @@ def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, .. manager = _register_all(build_hir(source)) registry = load_plugins(plugin_roots, manager) - manager.provide(SecurityModelAnalysis, plugin_models(registry)) + manager.provide(SecurityModelAnalysis, plugin_models(loaded.plugin for loaded in registry)) manager.provide(ProjectSummaries, SummaryIndex()) - return _check_module(manager, tuple(registry)) + return _check_module(manager, tuple(loaded.plugin for loaded in registry)) + + +def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph: + """The requirements declared or pinned by the dependency files at ``root`` (§26).""" + + graph = DependencyGraph() + candidates = sorted(root.glob("requirements*.txt")) + [root / name for name in DEPENDENCY_FILES] + for path in candidates: + if path.is_file(): + graph = graph.merge(parse_dependencies(sources.load_file(path))) + return graph -def analyze_project(root: Path, plugin_roots: Sequence[Path] = ()) -> ProjectAnalysis: - """Analyse every Python file under ``root`` with a shared summary index (§21).""" +def analyze_project( + root: Path, plugin_roots: Sequence[Path] = (), plugins: Sequence[Plugin] = () +) -> ProjectAnalysis: + """Analyse every Python file under ``root`` with a shared summary index (§21) and the + dependency graph of its manifests (§26). ``plugins`` adds plugin instances to the + ones discovered under ``plugin_roots``.""" sources = SourceManager() findings: list[Finding] = [] + dependencies = resolve_dependencies(root, sources) + for error in dependencies.errors: + findings.append(_note("syntax-error", error, sources.add_source(error.split(":")[0], ""), 1)) modules: dict[str, nodes.Module] = {} files: dict[str, SourceFile] = {} for source in discover_sources(root, sources): @@ -161,11 +194,14 @@ def analyze_project(root: Path, plugin_roots: Sequence[Path] = ()) -> ProjectAna managers = {name: _register_all(module) for name, module in modules.items()} probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("", ""))) registry = load_plugins(plugin_roots, probe) - models = plugin_models(registry) + all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *plugins) + models = plugin_models(all_plugins) + advisories = tuple(a for plugin in all_plugins for a in plugin.advisories) index = SummaryIndex() for manager in managers.values(): manager.provide(SecurityModelAnalysis, models) manager.provide(ProjectSummaries, index) + manager.provide(DependencyAnalysis, dependencies) imports: dict[str, ImportTable] = {} analysable: dict[str, AnalysisManager] = {} @@ -195,10 +231,14 @@ def analyze_project(root: Path, plugin_roots: Sequence[Path] = ()) -> ProjectAna manager.run(ProjectSummariesUpdated) manager.provide(ProjectSummaries, index) - plugins = tuple(registry) + module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin)) for name in sorted(analysable): - findings.extend(_check_module(analysable[name], plugins)) - return ProjectAnalysis(graph, index, tuple(findings)) + findings.extend(_check_module(analysable[name], module_plugins)) + context = ProjectContext(graph, dependencies, advisories, analysable) + for plugin in all_plugins: + if isinstance(plugin, ProjectPlugin): + findings.extend(plugin.analyze_project(context)) + return ProjectAnalysis(graph, index, tuple(findings), dependencies) def _summaries_of(manager: AnalysisManager) -> dict[str, FunctionSummary]: @@ -206,7 +246,7 @@ def _summaries_of(manager: AnalysisManager) -> dict[str, FunctionSummary]: return {name: table.summary(name) for name in table.names} -def _check_module(manager: AnalysisManager, plugins: tuple[LoadedPlugin, ...]) -> tuple[Finding, ...]: +def _check_module(manager: AnalysisManager, plugins: tuple[Plugin, ...]) -> tuple[Finding, ...]: supported: list[nodes.Function] = [] notes: list[Finding] = [] for function in analyzable_functions(manager.module): @@ -225,7 +265,7 @@ def _check_module(manager: AnalysisManager, plugins: tuple[LoadedPlugin, ...]) - ) else: supported.append(function) - findings = run_plugins(manager, [loaded.plugin for loaded in plugins], tuple(supported)) + findings = run_plugins(manager, plugins, tuple(supported)) return (*findings, *notes) diff --git a/src/coretrace_python/plugins/__init__.py b/src/coretrace_python/plugins/__init__.py index 2611e64..d1c49e4 100644 --- a/src/coretrace_python/plugins/__init__.py +++ b/src/coretrace_python/plugins/__init__.py @@ -5,6 +5,8 @@ ModelPlugin, Plugin, PluginContext, + ProjectContext, + ProjectPlugin, run_plugins, ) from coretrace_python.plugins.detectors import SymbolCallDetector, TaintDetector @@ -34,6 +36,8 @@ "PluginContext", "PluginManifest", "PluginRegistry", + "ProjectContext", + "ProjectPlugin", "SymbolCallDetector", "TaintDetector", "VersionRange", diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 94f9e86..c2efcd8 100644 --- a/src/coretrace_python/plugins/api.py +++ b/src/coretrace_python/plugins/api.py @@ -9,7 +9,7 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Mapping, Sequence from typing import Any, ClassVar, overload from coretrace_python.analysis import ( @@ -20,9 +20,12 @@ UndeclaredDependencyError, ) from coretrace_python.analysis.provider import R +from coretrace_python.dependency import Advisory, DependencyGraph from coretrace_python.findings import Finding from coretrace_python.hir import nodes +from coretrace_python.interprocedural import CallGraph, CallGraphAnalysis, ModuleGraph from coretrace_python.ir.lowering import analyzable_functions +from coretrace_python.semantic.imports import ImportAnalysis, ImportTable from coretrace_python.taint import Model PLUGIN_API_VERSION = 1 @@ -35,6 +38,7 @@ class Plugin(ABC): name: ClassVar[str] requires: ClassVar[frozenset[AnyAnalysis]] = frozenset() models: ClassVar[tuple[Model, ...]] = () + advisories: ClassVar[tuple[Advisory, ...]] = () @abstractmethod def analyze(self, ctx: PluginContext) -> Sequence[Finding]: @@ -42,12 +46,50 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: class ModelPlugin(Plugin): - """A plugin that only contributes security models (architecture §15 providers).""" + """A plugin that only contributes security models or advisories (§15 providers).""" def analyze(self, ctx: PluginContext) -> Sequence[Finding]: return () +class ProjectContext: + """What a project-scoped plugin sees: the module graph, the dependency graph, the + advisories every plugin contributed, and each module's imports and call graph.""" + + def __init__( + self, + graph: ModuleGraph, + dependencies: DependencyGraph, + advisories: tuple[Advisory, ...], + managers: Mapping[str, AnalysisManager], + ) -> None: + self.graph = graph + self.dependencies = dependencies + self.advisories = advisories + self._managers = managers + + @property + def modules(self) -> tuple[str, ...]: + return tuple(self._managers) + + def imports(self, module: str) -> ImportTable: + return self._managers[module].get(ImportAnalysis) + + def call_graph(self, module: str) -> CallGraph: + return self._managers[module].get(CallGraphAnalysis) + + +class ProjectPlugin(Plugin): + """A plugin that runs once per project rather than once per module (§26).""" + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + return () + + @abstractmethod + def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: + raise NotImplementedError + + class PluginContext: """What one plugin may see while analysing one module."""