diff --git a/README.md b/README.md index b008581..85e1cc7 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,25 @@ shipped `dependency/` plugins report a requirement that allows a vulnerable vers its line, and every call in the project to an API the advisory affects. When attacker-controlled data reaches such a call, the engine correlates the four facts into one critical `exploitable-vulnerability` finding, judged like any other flow. The shipped -advisory database is a small offline sample; a live OSV feed is future work. +advisory database is a small offline sample. A real feed stays offline too: +`--import-advisories SRC OUT` converts a public OSV dump (a JSON file, a directory of +them or a zip archive) into a local advisory file once, and a directory check reads +`advisories.json` at the project root plus the files passed with `--advisories`, the +local file winning over a plugin for the same advisory. OSV records name no affected +APIs, so imported advisories feed the requirement checks and the SBOM; add +`affected_symbols` by hand to a local entry and it feeds reachability and correlation +too. A `coretrace-policy.toml` at the root, or the file passed with `--policy`, denies +packages (`denied-dependency`), requires pins (`unpinned-dependency`) and lists accepted +advisories whose findings are dropped: + +```toml +[dependencies] +deny = ["pycrypto"] +require_pinned = true + +[advisories] +ignore = ["CVE-2020-1747"] +``` `--sbom PATH` writes a CycloneDX 1.5 bill of materials of the dependency graph, one component per requirement with its package URL, and the advisories affecting them as diff --git a/plugins/dependency/dependency_policy/dependency_policy.py b/plugins/dependency/dependency_policy/dependency_policy.py new file mode 100644 index 0000000..ba2929e --- /dev/null +++ b/plugins/dependency/dependency_policy/dependency_policy.py @@ -0,0 +1,46 @@ +"""Requirements the project's policy denies or leaves unpinned.""" + +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, Severity +from coretrace_python.plugins import ProjectContext, ProjectPlugin + + +class DependencyPolicyPlugin(ProjectPlugin): + name: ClassVar[str] = "dependency-policy" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({DependencyAnalysis}) + + def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: + findings: list[Finding] = [] + for requirement in ctx.dependencies.requirements: + metadata = {"package": requirement.name, "specifier": requirement.specifier} + if ctx.policy.denies(requirement.name): + findings.append( + Finding( + "denied-dependency", + f"{requirement.name} is denied by the dependency policy", + Severity.HIGH, + Confidence.HIGH, + requirement.span, + None, + metadata, + ) + ) + elif ctx.policy.require_pinned and requirement.pinned is None: + findings.append( + Finding( + "unpinned-dependency", + f"{requirement.name}{requirement.specifier} is not pinned to one version", + Severity.LOW, + Confidence.HIGH, + requirement.span, + None, + metadata, + ) + ) + return findings diff --git a/plugins/dependency/dependency_policy/plugin.toml b/plugins/dependency/dependency_policy/plugin.toml new file mode 100644 index 0000000..700fdf1 --- /dev/null +++ b/plugins/dependency/dependency_policy/plugin.toml @@ -0,0 +1,9 @@ +name = "dependency-policy" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["dependency.graph"] +provides = ["policy.dependencies"] + +[entrypoint] +module = "dependency_policy" +class = "DependencyPolicyPlugin" diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index 7d5a2f7..ffb265f 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -2,13 +2,14 @@ import argparse import sys +import zipfile from pathlib import Path from coretrace_python import __version__, engine from coretrace_python.analysis import AnalysisError from coretrace_python.cache import ProjectCache from coretrace_python.cfg import CFGError -from coretrace_python.dependency import render_sbom +from coretrace_python.dependency import dump_advisories, import_osv, read_osv, render_sbom from coretrace_python.frontend import HIRBuildError, ParseError, build_hir from coretrace_python.ir.lowering import LoweringError, lower_module from coretrace_python.ir.printer import format_module @@ -42,7 +43,9 @@ def build_parser() -> argparse.ArgumentParser: prog=engine.TOOL_NAME, description="Analyze Python source with CoreTrace.", ) - parser.add_argument("path", type=Path, help="Python source file, or a directory for --check") + parser.add_argument( + "path", type=Path, nargs="?", help="Python source file, or a directory for --check" + ) parser.add_argument( "--emit-ir", action="store_true", @@ -94,14 +97,51 @@ def build_parser() -> argparse.ArgumentParser: metavar="PATH", help="with --check on a directory, write a CycloneDX bill of materials to PATH", ) + parser.add_argument( + "--advisories", + action="append", + type=Path, + default=[], + metavar="FILE", + help="with --check on a directory, a local advisory file to read in addition to " + "advisories.json at the root (repeatable)", + ) + parser.add_argument( + "--policy", + type=Path, + default=None, + metavar="FILE", + help="with --check on a directory, the dependency policy to apply instead of " + "coretrace-policy.toml at the root", + ) + parser.add_argument( + "--import-advisories", + nargs=2, + type=Path, + default=None, + metavar=("SRC", "OUT"), + help="convert an OSV dump (a JSON file, a directory of them or a zip archive) " + "into the local advisory file OUT, then exit", + ) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) + if args.import_advisories is not None: + source, out = args.import_advisories + try: + out.write_text(dump_advisories(import_osv(read_osv(source))), encoding="utf-8") + except (OSError, ValueError, zipfile.BadZipFile) as error: + print(f"error: {error}", file=sys.stderr) + return EXIT_ERROR + return EXIT_CLEAN if not (args.emit_ir or args.check): print("error: no action selected; pass --emit-ir or --check", file=sys.stderr) return EXIT_ERROR + if args.path is None: + print("error: a path is required", file=sys.stderr) + return EXIT_ERROR if args.format is not None and not args.check: print("error: --format only applies to --check", file=sys.stderr) return EXIT_ERROR @@ -117,6 +157,12 @@ def main(argv: list[str] | None = None) -> int: if args.sbom is not None and not (args.check and args.path.is_dir()): print("error: --sbom only applies to --check on a directory", file=sys.stderr) return EXIT_ERROR + if args.advisories and not (args.check and args.path.is_dir()): + print("error: --advisories only applies to --check on a directory", file=sys.stderr) + return EXIT_ERROR + if args.policy is not None and not (args.check and args.path.is_dir()): + print("error: --policy only applies to --check on a directory", file=sys.stderr) + return EXIT_ERROR if args.path.is_dir() and args.emit_ir: print(f"error: {args.path} is a directory; --emit-ir needs a file", file=sys.stderr) @@ -130,7 +176,12 @@ def main(argv: list[str] | None = None) -> int: if args.path.is_dir(): cache = None if args.cache is None else ProjectCache(args.cache) analysis = engine.analyze_project( - args.path, args.plugins, cache=cache, jobs=args.jobs or 1 + args.path, + args.plugins, + cache=cache, + jobs=args.jobs or 1, + advisory_files=args.advisories, + policy_file=args.policy, ) findings = analysis.findings if args.sbom is not None: diff --git a/src/coretrace_python/dependency/__init__.py b/src/coretrace_python/dependency/__init__.py index 32b9252..2f7644d 100644 --- a/src/coretrace_python/dependency/__init__.py +++ b/src/coretrace_python/dependency/__init__.py @@ -1,5 +1,13 @@ """Dependency resolution and advisories (architecture §26).""" +from coretrace_python.dependency.advisories import ( + ADVISORY_FILE, + AdvisoryFileError, + dump_advisories, + import_osv, + load_advisories, + read_osv, +) from coretrace_python.dependency.graph import ( DEPENDENCY_FILES, Advisory, @@ -10,16 +18,27 @@ normalize, parse_dependencies, ) +from coretrace_python.dependency.policy import POLICY_FILE, Policy, apply_policy, load_policy from coretrace_python.dependency.sbom import render_sbom __all__ = [ + "ADVISORY_FILE", "DEPENDENCY_FILES", + "POLICY_FILE", "Advisory", + "AdvisoryFileError", "DependencyAnalysis", "DependencyGraph", + "Policy", "Requirement", "Version", + "apply_policy", + "dump_advisories", + "import_osv", + "load_advisories", + "load_policy", "normalize", "parse_dependencies", + "read_osv", "render_sbom", ] diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py new file mode 100644 index 0000000..42193cf --- /dev/null +++ b/src/coretrace_python/dependency/advisories.py @@ -0,0 +1,168 @@ +"""Local advisory files and the OSV import (architecture §26). + +The analysis never touches the network. ``import_osv`` converts the records of a public +OSV dump into ``Advisory`` values, keeping the PyPI ecosystem and turning each range of +events into a version specifier; ``dump_advisories`` writes them as a small JSON file +that a project keeps at its root as ``advisories.json`` or passes with ``--advisories``. +OSV records name no affected APIs, so imported advisories feed the requirement checks +and the SBOM; a file completed by hand with ``affected_symbols`` also feeds the +reachability and correlation checks. +""" + +from __future__ import annotations + +import json +import zipfile +from collections.abc import Iterable, Iterator, Mapping +from pathlib import Path +from typing import Any + +from coretrace_python.dependency.graph import Advisory, normalize +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId + +ADVISORY_FILE = "advisories.json" +ADVISORY_SCHEMA = 1 + +_SEVERITIES = { + "LOW": Severity.LOW, + "MODERATE": Severity.MEDIUM, + "MEDIUM": Severity.MEDIUM, + "HIGH": Severity.HIGH, + "CRITICAL": Severity.CRITICAL, +} + + +class AdvisoryFileError(Exception): + """An advisory or policy file could not be read.""" + + +# --------------------------------------------------------------------------- OSV + + +def read_osv(path: Path) -> Iterator[Mapping[str, Any]]: + """The records of an OSV dump: one JSON file (a record or a list), a directory of + JSON files, or a zip archive of them.""" + + if path.is_dir(): + for file in sorted(path.glob("*.json")): + yield from _records(json.loads(file.read_text(encoding="utf-8"))) + elif path.suffix == ".zip": + with zipfile.ZipFile(path) as archive: + for name in sorted(archive.namelist()): + if name.endswith(".json"): + yield from _records(json.loads(archive.read(name).decode("utf-8"))) + else: + yield from _records(json.loads(path.read_text(encoding="utf-8"))) + + +def _records(data: Any) -> Iterator[Mapping[str, Any]]: + if isinstance(data, list): + for item in data: + if isinstance(item, Mapping): + yield item + elif isinstance(data, Mapping): + yield data + + +def import_osv(records: Iterable[Mapping[str, Any]]) -> tuple[Advisory, ...]: + """The PyPI advisories of OSV records, one per affected range.""" + + advisories: list[Advisory] = [] + for record in records: + identifier = record.get("id") + if not isinstance(identifier, str): + continue + summary = str(record.get("summary") or "").strip() + if not summary: + lines = str(record.get("details") or "").strip().splitlines() + summary = lines[0] if lines else identifier + aliases = tuple(str(a) for a in record.get("aliases") or []) + severity = _severity(record) + for affected in record.get("affected") or []: + package = (affected.get("package") or {}) if isinstance(affected, Mapping) else {} + if str(package.get("ecosystem", "")).lower() != "pypi" or not package.get("name"): + continue + name = normalize(str(package["name"])) + for specifier in _specifiers(affected.get("ranges") or []): + advisories.append(Advisory(identifier, name, specifier, summary, severity, (), aliases)) + return tuple(advisories) + + +def _severity(record: Mapping[str, Any]) -> Severity: + specific = record.get("database_specific") or {} + label = specific.get("severity") if isinstance(specific, Mapping) else None + return _SEVERITIES.get(str(label).upper(), Severity.MEDIUM) + + +def _specifiers(ranges: Iterable[Mapping[str, Any]]) -> list[str]: + found: list[str] = [] + for entry in ranges: + if entry.get("type") not in ("ECOSYSTEM", "SEMVER"): + continue + introduced: str | None = None + for event in entry.get("events") or []: + if "introduced" in event: + introduced = str(event["introduced"]) + elif "fixed" in event or "last_affected" in event: + bound = f"<{event['fixed']}" if "fixed" in event else f"<={event['last_affected']}" + found.append(_clause(introduced, bound)) + introduced = None + if introduced is not None: + found.append(_clause(introduced, None)) + return found + + +def _clause(introduced: str | None, bound: str | None) -> str: + parts: list[str] = [] + if introduced is not None and introduced != "0": + parts.append(f">={introduced}") + if bound is not None: + parts.append(bound) + return ",".join(parts) if parts else ">=0" + + +# --------------------------------------------------------------------------- local file + + +def dump_advisories(advisories: Iterable[Advisory]) -> str: + document = { + "schema": ADVISORY_SCHEMA, + "advisories": [ + { + "id": a.id, + "package": a.package, + "vulnerable": a.vulnerable, + "summary": a.summary, + "severity": a.severity.value, + "affected_symbols": [str(s) for s in a.affected_symbols], + "aliases": list(a.aliases), + } + for a in advisories + ], + } + return json.dumps(document, indent=2) + "\n" + + +def load_advisories(path: Path) -> tuple[Advisory, ...]: + try: + document = json.loads(path.read_text(encoding="utf-8")) + if document.get("schema") != ADVISORY_SCHEMA: + raise AdvisoryFileError(f"{path}: unsupported advisory schema {document.get('schema')!r}") + return tuple(_advisory(entry) for entry in document["advisories"]) + except AdvisoryFileError: + raise + except (OSError, ValueError, KeyError, TypeError, AttributeError) as error: + raise AdvisoryFileError(f"{path}: {error}") from error + + +def _advisory(entry: Mapping[str, Any]) -> Advisory: + return Advisory( + str(entry["id"]), + normalize(str(entry["package"])), + str(entry["vulnerable"]), + str(entry["summary"]), + Severity(entry["severity"]), + tuple(SymbolId(str(s)) for s in entry.get("affected_symbols") or []), + tuple(str(a) for a in entry.get("aliases") or []), + ) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index a38bae6..963ce57 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -137,6 +137,7 @@ class Advisory: summary: str severity: Severity affected_symbols: tuple[SymbolId, ...] = () + aliases: tuple[str, ...] = () def affects(self, requirement: Requirement) -> bool: return requirement.name == normalize(self.package) and requirement.may_match(self.vulnerable) diff --git a/src/coretrace_python/dependency/policy.py b/src/coretrace_python/dependency/policy.py new file mode 100644 index 0000000..2b49aa7 --- /dev/null +++ b/src/coretrace_python/dependency/policy.py @@ -0,0 +1,65 @@ +"""Dependency policies (architecture §26). + +A ``coretrace-policy.toml`` at the project root, or the file passed with ``--policy``, +denies packages, requires pinned versions and lists the advisories a project accepts, +whose findings are dropped whatever produced them. +""" + +from __future__ import annotations + +import tomllib +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from coretrace_python.dependency.advisories import AdvisoryFileError +from coretrace_python.dependency.graph import normalize +from coretrace_python.findings import Finding + +POLICY_FILE = "coretrace-policy.toml" + + +@dataclass(frozen=True) +class Policy: + deny: tuple[str, ...] = () + require_pinned: bool = False + ignore: tuple[str, ...] = () + + def denies(self, package: str) -> bool: + return normalize(package) in {normalize(name) for name in self.deny} + + +def load_policy(path: Path) -> Policy: + try: + data = tomllib.loads(path.read_text(encoding="utf-8")) + dependencies = data.get("dependencies", {}) + advisories = data.get("advisories", {}) + return Policy( + _strings(dependencies.get("deny", []), "dependencies.deny"), + _boolean(dependencies.get("require_pinned", False), "dependencies.require_pinned"), + _strings(advisories.get("ignore", []), "advisories.ignore"), + ) + except AdvisoryFileError as error: + raise AdvisoryFileError(f"{path}: {error}") from error + except (OSError, tomllib.TOMLDecodeError, AttributeError) as error: + raise AdvisoryFileError(f"{path}: {error}") from error + + +def _strings(value: Any, key: str) -> tuple[str, ...]: + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise AdvisoryFileError(f"{key} must be a list of strings") + return tuple(value) + + +def _boolean(value: Any, key: str) -> bool: + if not isinstance(value, bool): + raise AdvisoryFileError(f"{key} must be true or false") + return value + + +def apply_policy(policy: Policy, findings: Iterable[Finding]) -> tuple[Finding, ...]: + """The findings minus those about an advisory the policy accepts.""" + + ignored = set(policy.ignore) + return tuple(f for f in findings if f.metadata.get("advisory") not in ignored) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index e94b4e3..cd838a6 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -35,10 +35,17 @@ ) from coretrace_python.cfg import CFGAnalysis, CFGError, DominanceAnalysis, PostDominanceAnalysis from coretrace_python.dependency import ( + ADVISORY_FILE, DEPENDENCY_FILES, + POLICY_FILE, Advisory, + AdvisoryFileError, DependencyAnalysis, DependencyGraph, + Policy, + apply_policy, + load_advisories, + load_policy, parse_dependencies, ) from coretrace_python.dependency.correlation import advisory_sinks, affected_symbols, correlate @@ -217,13 +224,17 @@ def analyze_project( plugins: Sequence[Plugin] = (), cache: ProjectCache | None = None, jobs: int = 1, + advisory_files: Sequence[Path] = (), + policy_file: Path | None = None, ) -> 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``. With a ``cache``, modules whose key is unchanged since a previous run are served from it (§11). Modules are scheduled by strongly connected components of the module graph, imports first; with ``jobs`` - above one the components of a wave are analysed in that many processes (§29).""" + above one the components of a wave are analysed in that many processes (§29). + ``advisory_files`` add to the ``advisories.json`` at ``root``; ``policy_file`` + replaces the ``coretrace-policy.toml`` there.""" if jobs < 1: raise ValueError("jobs must be at least 1") @@ -232,6 +243,20 @@ def analyze_project( dependencies = resolve_dependencies(root, sources) for error in dependencies.errors: findings.append(_note("syntax-error", error, sources.add_source(error.split(":")[0], ""), 1)) + advisory_paths = _advisory_paths(root, advisory_files) + file_advisories: list[Advisory] = [] + for path in advisory_paths: + try: + file_advisories.extend(load_advisories(path)) + except AdvisoryFileError as error: + findings.append(_note("syntax-error", str(error), sources.add_source(str(path), ""), 1)) + policy = Policy() + policy_path = policy_file if policy_file is not None else root / POLICY_FILE + if policy_path.is_file(): + try: + policy = load_policy(policy_path) + except AdvisoryFileError as error: + findings.append(_note("syntax-error", str(error), sources.add_source(str(policy_path), ""), 1)) modules: dict[str, nodes.Module] = {} files: dict[str, SourceFile] = {} for source in discover_sources(root, sources): @@ -246,7 +271,9 @@ def analyze_project( probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("", ""))) registry = load_plugins(plugin_roots, probe) all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *plugins) - advisories = tuple(a for plugin in all_plugins for a in plugin.advisories) + advisories = _merge_advisories( + (a for plugin in all_plugins for a in plugin.advisories), file_advisories + ) affected = affected_symbols(dependencies, advisories) models = plugin_models(all_plugins).extended(*advisory_sinks(affected)) for manager in managers.values(): @@ -305,6 +332,7 @@ def analyze_project( tuple(plugins), {name: _path_of(files[name]) for name in sorted(component)}, encode_index(_seed(results, graph, component)), + advisory_paths, ), ) for component in pending @@ -328,15 +356,40 @@ def analyze_project( for site in entry.sites: sites.setdefault(site.caller, []).append(site) call_graphs[name] = CallGraph({}, {f: tuple(s) for f, s in sites.items()}, frozenset()) - context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs) + context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs, policy) for plugin in all_plugins: if isinstance(plugin, ProjectPlugin): findings.extend(plugin.analyze_project(context)) return ProjectAnalysis( - graph, index, tuple(findings), dependencies, MappingProxyType(keys), reused, advisories + graph, + index, + apply_policy(policy, findings), + dependencies, + MappingProxyType(keys), + reused, + advisories, ) +def _merge_advisories( + from_plugins: Iterable[Advisory], from_files: Iterable[Advisory] +) -> tuple[Advisory, ...]: + """One advisory per identifier, package and range; a local file's version wins over + a plugin's, since the file is the project's own curated feed.""" + + merged: dict[tuple[str, str, str], Advisory] = {} + for advisory in (*from_plugins, *from_files): + merged[(advisory.id, advisory.package, advisory.vulnerable)] = advisory + return tuple(merged.values()) + + +def _advisory_paths(root: Path, advisory_files: Sequence[Path]) -> tuple[Path, ...]: + """The advisory file at ``root``, when present, then the explicit ones.""" + + default = root / ADVISORY_FILE + return (*([default] if default.is_file() else []), *advisory_files) + + def _seed(results: Mapping[str, CachedModule], graph: ModuleGraph, component: frozenset[str]) -> SummaryIndex: """The summaries a component starts from: those of every module it imports, transitively, all final by the time its wave runs; the whole index for no component.""" @@ -402,6 +455,7 @@ class _Batch: plugins: tuple[Plugin, ...] paths: Mapping[str, Path] seed: Mapping[str, Any] + advisory_paths: tuple[Path, ...] = () def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: @@ -410,7 +464,15 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: registry = load_plugins(batch.plugin_roots, next(iter(managers.values()))) all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *batch.plugins) dependencies = resolve_dependencies(batch.root, sources) - advisories = tuple(a for plugin in all_plugins for a in plugin.advisories) + file_advisories: list[Advisory] = [] + for path in batch.advisory_paths: + try: + file_advisories.extend(load_advisories(path)) + except AdvisoryFileError: + continue + advisories = _merge_advisories( + (a for plugin in all_plugins for a in plugin.advisories), file_advisories + ) affected = affected_symbols(dependencies, advisories) models = plugin_models(all_plugins).extended(*advisory_sinks(affected)) for manager in managers.values(): diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 4989524..5d20413 100644 --- a/src/coretrace_python/plugins/api.py +++ b/src/coretrace_python/plugins/api.py @@ -20,7 +20,7 @@ UndeclaredDependencyError, ) from coretrace_python.analysis.provider import R -from coretrace_python.dependency import Advisory, DependencyGraph +from coretrace_python.dependency import Advisory, DependencyGraph, Policy from coretrace_python.findings import Finding from coretrace_python.hir import nodes from coretrace_python.interprocedural import CallGraph, CallGraphAnalysis, ModuleGraph @@ -54,8 +54,9 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: 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. The - engine passes the call graphs of modules it served from its cache.""" + advisories every plugin and advisory file contributed, the dependency policy, and + each module's imports and call graph. The engine passes the call graphs of modules + it served from its cache.""" def __init__( self, @@ -64,10 +65,12 @@ def __init__( advisories: tuple[Advisory, ...], managers: Mapping[str, AnalysisManager], call_graphs: Mapping[str, CallGraph] | None = None, + policy: Policy | None = None, ) -> None: self.graph = graph self.dependencies = dependencies self.advisories = advisories + self.policy = policy or Policy() self._managers = managers self._call_graphs = dict(call_graphs or {}) diff --git a/tests/test_advisory_feeds.py b/tests/test_advisory_feeds.py new file mode 100644 index 0000000..535c938 --- /dev/null +++ b/tests/test_advisory_feeds.py @@ -0,0 +1,360 @@ +"""Acceptance tests for local advisory feeds and dependency policies (``docs/architecture.md`` +§26; roadmap issue #37, last point). + +The analysis stays offline and deterministic: ``--import-advisories SRC OUT`` converts a +public OSV dump (a JSON file, a directory of them or a zip archive) into a local advisory +file once, and a directory check reads ``advisories.json`` at the project root or the +files passed with ``--advisories``. A ``coretrace-policy.toml`` at the root, or the file +passed with ``--policy``, denies packages, requires pins and accepts advisories. + +Expected to remain red until ``coretrace_python.dependency.advisories``, +``coretrace_python.dependency.policy``, the ``dependency-policy`` plugin and the CLI +options exist. +""" + +from __future__ import annotations + +import json +import zipfile +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.dependency import Advisory +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.semantic.symbols import SymbolId + +try: + from coretrace_python.dependency.advisories import ( + AdvisoryFileError, + dump_advisories, + import_osv, + load_advisories, + read_osv, + ) + from coretrace_python.dependency.policy import Policy, apply_policy, load_policy +except ImportError as error: # pragma: no cover - red until feeds and policies land + MISSING: Exception | None = error +else: + MISSING = None + if "aliases" not in Advisory.__dataclass_fields__: + MISSING = AttributeError("Advisory has no aliases") + + +@pytest.fixture(autouse=True) +def require_feeds() -> None: + if MISSING is not None: + pytest.fail(f"advisory feeds are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def osv(identifier: str, package: str, events: list[dict[str, str]], **extra: object) -> dict[str, object]: + return { + "id": identifier, + "summary": f"{identifier} summary", + "details": "Long details.\nSecond line.", + "aliases": ["CVE-2000-0001"], + "affected": [ + { + "package": {"ecosystem": "PyPI", "name": package}, + "ranges": [{"type": "ECOSYSTEM", "events": events}], + } + ], + **extra, + } + + +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 rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: + return sorted((Path(str(f.span.source_id)).name, f.rule_id, f.span.start_line) for f in findings) + + +# --------------------------------------------------------------------------- OSV import + + +def test_osv_ranges_become_version_specifiers() -> None: + advisories = import_osv( + [ + osv("GHSA-1", "PyYAML", [{"introduced": "0"}, {"fixed": "5.4"}]), + osv("GHSA-2", "requests", [{"introduced": "1.0"}, {"fixed": "2.0"}]), + osv("GHSA-3", "flask", [{"introduced": "1.0"}, {"last_affected": "1.5"}]), + osv("GHSA-4", "django", [{"introduced": "3.0"}]), + ] + ) + + assert [(a.id, a.package, a.vulnerable) for a in advisories] == [ + ("GHSA-1", "pyyaml", "<5.4"), + ("GHSA-2", "requests", ">=1.0,<2.0"), + ("GHSA-3", "flask", ">=1.0,<=1.5"), + ("GHSA-4", "django", ">=3.0"), + ] + assert advisories[0].summary == "GHSA-1 summary" + assert advisories[0].aliases == ("CVE-2000-0001",) + assert advisories[0].affected_symbols == () + + +def test_osv_severity_summary_and_ecosystem_handling() -> None: + critical = osv("GHSA-5", "a", [{"introduced": "0"}], database_specific={"severity": "CRITICAL"}) + moderate = osv("GHSA-6", "b", [{"introduced": "0"}], database_specific={"severity": "MODERATE"}) + unknown = osv("GHSA-7", "c", [{"introduced": "0"}]) + unknown["summary"] = "" + npm = osv("GHSA-8", "d", [{"introduced": "0"}]) + npm["affected"][0]["package"]["ecosystem"] = "npm" # type: ignore[index] + two_ranges = osv("GHSA-9", "e", [{"introduced": "0"}, {"fixed": "1.1"}]) + two_ranges["affected"][0]["ranges"].append( # type: ignore[index] + {"type": "ECOSYSTEM", "events": [{"introduced": "2.0"}, {"fixed": "2.1"}]} + ) + + advisories = import_osv([critical, moderate, unknown, npm, two_ranges]) + + assert [(a.id, a.severity) for a in advisories] == [ + ("GHSA-5", Severity.CRITICAL), + ("GHSA-6", Severity.MEDIUM), + ("GHSA-7", Severity.MEDIUM), + ("GHSA-9", Severity.MEDIUM), + ("GHSA-9", Severity.MEDIUM), + ] + assert advisories[2].summary == "Long details." + assert [a.vulnerable for a in advisories if a.id == "GHSA-9"] == ["<1.1", ">=2.0,<2.1"] + + +def test_osv_dumps_are_read_from_files_directories_and_archives(tmp_path: Path) -> None: + record = osv("GHSA-1", "pyyaml", [{"introduced": "0"}, {"fixed": "5.4"}]) + single = tmp_path / "one.json" + single.write_text(json.dumps(record), encoding="utf-8") + listed = tmp_path / "list.json" + listed.write_text(json.dumps([record, record | {"id": "GHSA-2"}]), encoding="utf-8") + folder = tmp_path / "osv" + folder.mkdir() + (folder / "a.json").write_text(json.dumps(record | {"id": "GHSA-3"}), encoding="utf-8") + (folder / "b.json").write_text(json.dumps(record | {"id": "GHSA-4"}), encoding="utf-8") + (folder / "notes.txt").write_text("ignored", encoding="utf-8") + archive = tmp_path / "all.zip" + with zipfile.ZipFile(archive, "w") as bundle: + bundle.writestr("GHSA-5.json", json.dumps(record | {"id": "GHSA-5"})) + bundle.writestr("README", "ignored") + + assert [r["id"] for r in read_osv(single)] == ["GHSA-1"] + assert [r["id"] for r in read_osv(listed)] == ["GHSA-1", "GHSA-2"] + assert [r["id"] for r in read_osv(folder)] == ["GHSA-3", "GHSA-4"] + assert [r["id"] for r in read_osv(archive)] == ["GHSA-5"] + + +# --------------------------------------------------------------------------- local file + + +ADVISORY = ( + Advisory( + "CVE-2020-1747", + "pyyaml", + "<5.4", + "unsafe load", + Severity.CRITICAL, + (SymbolId("python.yaml.load"),), + ("GHSA-8q59-q68h-6hv4",), + ) + if MISSING is None + else None +) + + +def test_advisory_files_round_trip(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + + assert load_advisories(path) == (ADVISORY,) + document = json.loads(path.read_text(encoding="utf-8")) + assert document["schema"] == 1 + assert document["advisories"][0]["affected_symbols"] == ["python.yaml.load"] + + +def test_malformed_advisory_files_are_reported(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text('{"schema": 1, "advisories": [{"id": "x"}]}', encoding="utf-8") + with pytest.raises(AdvisoryFileError): + load_advisories(path) + path.write_text("{not json", encoding="utf-8") + with pytest.raises(AdvisoryFileError): + load_advisories(path) + + +def test_the_cli_imports_an_osv_dump(tmp_path: Path) -> None: + source = tmp_path / "osv.json" + source.write_text(json.dumps([osv("GHSA-1", "PyYAML", [{"introduced": "0"}, {"fixed": "5.4"}])]), encoding="utf-8") + out = tmp_path / "advisories.json" + + assert main(["--import-advisories", str(source), str(out)]) == 0 + + (advisory,) = load_advisories(out) + assert (advisory.id, advisory.package, advisory.vulnerable) == ("GHSA-1", "pyyaml", "<5.4") + + +# --------------------------------------------------------------------------- feeds in a check + + +def test_projects_read_advisories_json_at_their_root(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "left-pad==1.0\n", + "app.py": "x = 1\n", + "advisories.json": dump_advisories( + (Advisory("GHSA-LP", "left-pad", "<1.1", "padding overflow", Severity.HIGH),) + ), + }, + ) + + analysis = engine.analyze_project(root, [PLUGINS]) + + assert rules(analysis.findings) == [("requirements.txt", "vulnerable-dependency", 1)] + assert any(a.id == "GHSA-LP" for a in analysis.advisories) + + +def test_advisory_files_are_passed_explicitly_and_feed_the_correlation(tmp_path: Path) -> None: + feed = tmp_path / "feed.json" + feed.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + root = project( + tmp_path / "src", + {"requirements.txt": "pyyaml==5.3.1\n", "app.py": "import yaml\n\ndef load():\n return yaml.load(input())\n"}, + ) + + findings = engine.analyze_project(root, [PLUGINS], advisory_files=[feed]).findings + + assert rules(findings) == [ + ("app.py", "exploitable-vulnerability", 4), + ("app.py", "reachable-vulnerability", 4), + ("requirements.txt", "vulnerable-dependency", 1), + ] + + +def test_malformed_advisory_files_become_notes(tmp_path: Path) -> None: + root = project(tmp_path, {"advisories.json": "{not json", "app.py": "x = 1\n"}) + + (note,) = engine.analyze_project(root, [PLUGINS]).findings + + assert (note.rule_id, Path(str(note.span.source_id)).name) == ("syntax-error", "advisories.json") + + +def test_cli_accepts_advisory_files(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + feed = tmp_path / "feed.json" + feed.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + root = project(tmp_path / "src", {"requirements.txt": "pyyaml==5.3.1\n", "app.py": "x = 1\n"}) + + assert main(["--check", str(root), "--plugins", str(PLUGINS), "--advisories", str(feed)]) == 1 + assert "vulnerable-dependency" in capsys.readouterr().out + assert main(["--emit-ir", "--advisories", str(feed), str(root / "app.py")]) == 2 + assert "--advisories" in capsys.readouterr().err + + +# --------------------------------------------------------------------------- policies + + +POLICY = ( + "[dependencies]\n" + 'deny = ["pycrypto", "left-pad"]\n' + "require_pinned = true\n\n" + "[advisories]\n" + 'ignore = ["CVE-2020-1747"]\n' +) + + +def test_policies_load_from_toml(tmp_path: Path) -> None: + path = tmp_path / "coretrace-policy.toml" + path.write_text(POLICY, encoding="utf-8") + + policy = load_policy(path) + + assert policy == Policy(deny=("pycrypto", "left-pad"), require_pinned=True, ignore=("CVE-2020-1747",)) + assert Policy() == Policy(deny=(), require_pinned=False, ignore=()) + path.write_text("[dependencies]\ndeny = 3\n", encoding="utf-8") + with pytest.raises(AdvisoryFileError): + load_policy(path) + + +def test_accepted_advisories_drop_their_findings() -> None: + from coretrace_python.source import SourceId, SourceSpan + + span = SourceSpan(SourceId("r.txt"), 1, 1) + kept = Finding("vulnerable-dependency", "m", Severity.HIGH, Confidence.HIGH, span, None, {"advisory": "GHSA-X"}) + dropped = Finding("exploitable-vulnerability", "m", Severity.HIGH, Confidence.HIGH, span, "f", {"advisory": "CVE-2020-1747"}) + other = Finding("command-injection", "m", Severity.HIGH, Confidence.HIGH, span, "f") + + assert apply_policy(Policy(ignore=("CVE-2020-1747",)), (kept, dropped, other)) == (kept, other) + + +def test_denied_and_unpinned_requirements_are_reported(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "pycrypto==2.6.1\nrequests>=2.20\nflask==3.0.0\n", + "coretrace-policy.toml": POLICY, + "app.py": "x = 1\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [ + ("requirements.txt", "denied-dependency", 1), + ("requirements.txt", "unpinned-dependency", 2), + ] + denied = next(f for f in findings if f.rule_id == "denied-dependency") + assert denied.severity is Severity.HIGH and denied.metadata["package"] == "pycrypto" + unpinned = next(f for f in findings if f.rule_id == "unpinned-dependency") + assert unpinned.severity is Severity.LOW and unpinned.metadata["specifier"] == ">=2.20" + + +def test_policies_silence_accepted_advisories_in_a_project(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "pyyaml==5.3.1\n", + "coretrace-policy.toml": '[advisories]\nignore = ["CVE-2020-1747"]\n', + "app.py": "import yaml\n\ndef load():\n return yaml.load(input())\n", + }, + ) + + assert engine.analyze_project(root, [PLUGINS]).findings == () + + +def test_policy_plugin_is_shipped_and_sees_the_policy() -> None: + from coretrace_python.frontend import build_hir + from coretrace_python.plugins import ProjectContext, discover_plugins + from coretrace_python.source import SourceManager + + module = build_hir(SourceManager().add_source("empty.py", "")) + loaded = {p.manifest.name: p.manifest for p in discover_plugins(PLUGINS, engine.build_manager(module))} + + assert loaded["dependency-policy"].provides == ("policy.dependencies",) + assert "policy" in ProjectContext.__init__.__code__.co_varnames + + +def test_cli_accepts_a_policy_file(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + policy = tmp_path / "policy.toml" + policy.write_text('[dependencies]\ndeny = ["flask"]\n', encoding="utf-8") + root = project(tmp_path / "src", {"requirements.txt": "flask==3.0.0\n", "app.py": "x = 1\n"}) + + assert main(["--check", str(root), "--plugins", str(PLUGINS), "--policy", str(policy)]) == 1 + assert "denied-dependency" in capsys.readouterr().out + assert main(["--check", "--policy", str(policy), str(root / "app.py")]) == 2 + assert "--policy" in capsys.readouterr().err + + +def test_malformed_policies_become_notes(tmp_path: Path) -> None: + root = project(tmp_path, {"coretrace-policy.toml": "deny = [", "app.py": "x = 1\n"}) + + (note,) = engine.analyze_project(root, [PLUGINS]).findings + + assert (note.rule_id, Path(str(note.span.source_id)).name) == ("syntax-error", "coretrace-policy.toml") diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 220bb91..92bee2c 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -164,6 +164,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: assert set(by_name) == { "command-injection", "dangerous-eval", + "dependency-policy", "django-models", "fastapi-models", "flask-models",