Skip to content
Merged
8 changes: 6 additions & 2 deletions docs/plugins.md
Original file line number Diff line number Diff line change
Expand Up @@ -320,8 +320,12 @@ any analysis result: `SSAAnalysis` (`coretrace_python.ir.ssa`) for the code,
## Findings

A `Finding` (`coretrace_python.findings`) has a `rule_id`, a `message`, a `Severity`, a
`Confidence`, a `SourceSpan` (file, start and end line and column), the enclosing
`function` and string `metadata`. Rule ids are lower-case words joined by dashes.
`Confidence`, a location, the enclosing `function` and string `metadata`. Rule ids are
lower-case words joined by dashes. The location (`span`) is a `SourceSpan` (file, start
and end line and column) when the line is established, or a `FileLocation` (file and the
JSON pointer of a value) for a value of a structured file whose line cannot be: a plugin
that reads a finding's line checks which it has, and never invents a line for a
`FileLocation` (`coretrace_python.source.positions` places JSON and TOML values).

## Testing a plugin

Expand Down
25 changes: 23 additions & 2 deletions docs/usage.md
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,8 @@ The comment also works in requirements files, where a line
`pyyaml==5.3.1 # coretrace: ignore[vulnerable-dependency]` silences that requirement's
finding. Suppressed findings are kept apart: the text report counts them, the JSON report
lists them under `suppressed`, the SARIF log marks them as suppressed in source, and
they never affect the exit status.
they never affect the exit status. A comment only suppresses a finding located at a
verified line; a finding with a file location (see Reports) is never suppressed this way.

## Adopting the analyzer on an existing code base

Expand All @@ -232,7 +233,16 @@ above it does not make it new; a change to the line itself does. Commit the file
shrink it as findings are fixed: an entry without a matching finding is simply unused.
Baselined findings are counted in the text report, listed under `baselined` in the JSON
report and marked `baselineState: unchanged` in the SARIF log, where new results are
marked `new`.
marked `new`. A finding with a file location is recognised by the JSON pointer of its
value and a digest of the value (never the value itself) instead of the text of a line,
so a new value at the same place is a new finding.

Baselines are written in schema 2. A schema 1 file, written by an earlier version, is
still read and its entries match exactly as they did; but it recorded a finding in a
JSON, TOML or lock file with the text of its key's first line, which was its own only
when the key was not repeated. A finding such an entry cannot identify with certainty is
reported as new, and the check then warns on standard error: review the new findings,
delete the baseline and run the check again to record it in schema 2.

## Reports

Expand Down Expand Up @@ -267,6 +277,17 @@ the directory of the checked file; a path outside it is printed as it is.
}
```

Lines are counted as Python counts them, by `\n`, `\r\n` or a lone `\r` (never U+2028
or a form feed), in every report, suppression and baseline. A value of a JSON or TOML
file, a package of a lock file, or a requirement of `pyproject.toml`, is located at its own line,
which a separate component establishes from the value's structural path and checks
against the value. When it cannot (a dotted TOML key, an inline table, a multi-line
value), the finding gets a file location instead of a guessed line: `line`, `column`,
`end_line` and `end_column` are `null`, and `pointer` gives the JSON pointer of the value
in the file (`"/tool/app/api_token"`). The SARIF log gives such a result no `region`,
with the pointer under the location's `properties`, and the text report shows the file
alone (`settings.toml: high hardcoded-credential: …`).

`root` is the directory the paths are relative to. A file's `status` is `analysed`,
`ambiguous` (analysed, but another file has its module name, see above), `syntax-error`
or `unreadable`; only `analysed` files count in `files_analysed`.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,11 @@
from __future__ import annotations

from collections.abc import Sequence
from pathlib import PurePath
from typing import ClassVar

from coretrace_python.analysis import AnyAnalysis
from coretrace_python.dependency import DependencyAnalysis
from coretrace_python.dependency import Advisory, DependencyAnalysis, Requirement
from coretrace_python.dependency.correlation import (
affected_entries,
affected_symbols,
Expand Down Expand Up @@ -44,11 +45,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
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}"
),
message=_message(ctx, advisory, requirement),
severity=advisory.severity,
confidence=Confidence.HIGH if pinned else Confidence.MEDIUM,
span=requirement.span,
Expand All @@ -58,6 +55,26 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
return findings


def _message(ctx: ProjectContext, advisory: Advisory, requirement: Requirement) -> str:
"""The evidence: for a locked package, the version the lock file pins and the
packages that require it; for a declared requirement, the specifier that allows or
requires a vulnerable version."""

if requirement.locked is None:
verb = "required" if requirement.pinned is not None else "allowed"
return (
f"{advisory.id}: {advisory.package} {advisory.vulnerable} is {verb} by "
f"{requirement.name}{requirement.specifier or ''}: {advisory.summary}"
)
lock = PurePath(str(requirement.span.source_id)).name
dependents = sorted(ctx.dependencies.required_by(requirement.name) or ())
required = f" (required by {', '.join(dependents)})" if dependents else ""
return (
f"{advisory.id}: {advisory.package} {requirement.locked}, pinned in {lock}, "
f"is in {advisory.vulnerable}{required}: {advisory.summary}"
)


def _ruled_out(ctx: ProjectContext) -> dict[str, list[str]]:
"""By advisory, the calls to its entry points whose arguments contradict one of its
conditions: the evidence that the requirement, imported, is not reached there."""
Expand Down
26 changes: 22 additions & 4 deletions src/coretrace_python/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,9 @@
UnknownTarget,
)
from coretrace_python.semantic.symbols import SymbolId
from coretrace_python.source import SourceId, SourceSpan
from coretrace_python.source import FileLocation, Location, SourceId, SourceSpan

CACHE_FORMAT = 13
CACHE_FORMAT = 14


@dataclass(frozen=True)
Expand Down Expand Up @@ -185,6 +185,24 @@ def _encode_span(span: SourceSpan) -> list[Any]:
return [str(span.source_id), span.start_line, span.start_column, span.end_line, span.end_column]


def _encode_location(location: Location) -> Any:
if isinstance(location, FileLocation):
return {
"file": str(location.source_id),
"pointer": location.pointer,
"digest": location.digest,
}
return _encode_span(location)


def _decode_location(data: Any) -> Location:
if isinstance(data, Mapping):
return FileLocation(
SourceId(_string(data["file"])), _string(data["pointer"]), _string(data["digest"])
)
return _decode_span(data)


def _decode_span(data: Any) -> SourceSpan:
file, line, column, end_line, end_column = data
return SourceSpan(
Expand Down Expand Up @@ -217,7 +235,7 @@ def _encode_finding(finding: Finding) -> dict[str, Any]:
"message": finding.message,
"severity": finding.severity.value,
"confidence": finding.confidence.value,
"span": _encode_span(finding.span),
"span": _encode_location(finding.span),
"function": finding.function,
"metadata": dict(finding.metadata),
}
Expand All @@ -230,7 +248,7 @@ def _decode_finding(data: Mapping[str, Any]) -> Finding:
_string(data["message"]),
Severity(data["severity"]),
Confidence(data["confidence"]),
_decode_span(data["span"]),
_decode_location(data["span"]),
None if function is None else _string(function),
{_string(k): _string(v) for k, v in data["metadata"].items()},
)
Expand Down
6 changes: 5 additions & 1 deletion src/coretrace_python/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -274,7 +274,11 @@ def main(argv: list[str] | None = None) -> int:
baselined = None
if args.baseline is not None:
if args.baseline.is_file():
findings, baselined = Baseline.load(args.baseline).partition(findings, root)
baseline = Baseline.load(args.baseline)
findings, baselined = baseline.partition(findings, root)
notice = baseline.transition_notice(findings)
if notice is not None:
print(f"coretrace: {args.baseline}: {notice}", file=sys.stderr)
else:
Baseline.of(findings, root).save(args.baseline)
findings, baselined = (), findings
Expand Down
81 changes: 51 additions & 30 deletions src/coretrace_python/dependency/graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,16 @@
import re
import tomllib
from collections.abc import Iterable, Iterator, Mapping
from dataclasses import dataclass
from dataclasses import dataclass, replace
from pathlib import PurePath, PurePosixPath
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
from coretrace_python.source import Location, SourceFile, SourceId, SourceSpan, lines_of
from coretrace_python.source.positions import Pointer, TomlPositions, file_location

_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.*+!-]*)$")
Expand Down Expand Up @@ -115,11 +116,16 @@ def _lower_bounds(specifier: str) -> list[Version]:

@dataclass(frozen=True)
class Requirement:
"""``span`` is where the requirement is written: a verified line, or, for a locked
package whose entry cannot be placed, a file location. ``locked`` is the version a
lock file pins, as it writes it; None for a declared requirement."""

name: str
specifier: str
span: SourceSpan
span: Location
pinned: Version | None = None
optional: bool = False
locked: str | None = None

@classmethod
def parse(cls, text: str, source_id: SourceId, line: int, optional: bool = False) -> Requirement | None:
Expand Down Expand Up @@ -313,12 +319,14 @@ def merge(self, other: DependencyGraph) -> DependencyGraph:
if current is None:
merged[name] = requirement
continue
declared = current if current.specifier else requirement
merged[name] = Requirement(
name,
current.specifier or requirement.specifier,
current.span if current.specifier else requirement.span,
declared.span,
requirement.pinned or current.pinned,
current.optional and requirement.optional,
declared.locked,
)
dependents = dict(self._dependents)
for name, found in other._dependents.items():
Expand All @@ -341,7 +349,7 @@ def parse_dependencies(source: SourceFile) -> DependencyGraph:

def _parse_requirements_txt(source: SourceFile) -> DependencyGraph:
found: dict[str, Requirement] = {}
for number, line in enumerate(source.text.splitlines(), start=1):
for number, line in enumerate(lines_of(source.text), start=1):
stripped = line.strip()
if not stripped or stripped.startswith(("#", "-")):
continue
Expand All @@ -362,25 +370,32 @@ def _parse_toml(source: SourceFile, extract: Any, dependents: Any = None) -> Dep
return DependencyGraph(found, dependents=None if dependents is None else dependents(data))


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]:
"""The requirements ``pyproject.toml`` declares (PEP 621 and Poetry), each located at
its own array element or key, which the position component checks, or given a file
location when it cannot be placed."""

positions = TomlPositions(source.text)

def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]:
found: list[Requirement] = []
def declared(text: str, path: Pointer, value: object, optional: bool = False) -> Requirement | None:
requirement = Requirement.parse(text, source.source_id, 1, optional)
if requirement is None:
return None
place = positions.locate(path, value)
location: Location = (
SourceSpan(source.source_id, *place)
if place is not None
else file_location(source.source_id, path, text)
)
return replace(requirement, span=location)

found: list[Requirement | None] = []
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)
for index, text in enumerate(project.get("dependencies", []) or []):
found.append(declared(text, ("project", "dependencies", index), text))
for group, texts in (project.get("optional-dependencies", {}) or {}).items():
for index, text in enumerate(texts or []):
found.append(declared(text, ("project", "optional-dependencies", group, index), text, True))
poetry = data.get("tool", {}).get("poetry", {})
for section, optional in (("dependencies", False), ("dev-dependencies", True)):
for name, spec in (poetry.get(section, {}) or {}).items():
Expand All @@ -389,23 +404,29 @@ def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list
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
found.append(declared(f"{name}{specifier}", ("tool", "poetry", section, name), spec, optional))
return [requirement for requirement in found if requirement is not None]


def _lock_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]:
"""The packages a lock file pins, each located at the ``name`` of its own
``[[package]]`` entry, which the position component checks."""

found: list[Requirement] = []
for package in data.get("package", []) or []:
positions = TomlPositions(source.text)
for index, package in enumerate(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}"')
path = ("package", index, "name")
place = positions.locate(path, name)
location: Location = (
SourceSpan(source.source_id, *place)
if place is not None
else file_location(source.source_id, path, name)
)
found.append(
Requirement(normalize(name), "", SourceSpan(source.source_id, line, 1), Version.parse(version))
Requirement(normalize(name), "", location, Version.parse(version), locked=version)
)
return found

Expand Down
7 changes: 5 additions & 2 deletions src/coretrace_python/dependency/vex.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from coretrace_python.dependency.sbom import purl
from coretrace_python.findings import Component, Finding
from coretrace_python.findings.coverage import Coverage
from coretrace_python.source import SourceSpan
from coretrace_python.taint import FILTER_FUNCTIONS

CONTEXT = "https://openvex.dev/ns/v0.2.0"
Expand Down Expand Up @@ -179,10 +180,12 @@ def _reached(findings: Sequence[Finding], root: Path) -> str:
level = finding.metadata.get("level")
if level in REACHED:
path = _relative(str(finding.span.source_id), root)
place = (path, finding.span.start_line, str(finding.metadata.get("symbol")))
line = finding.span.start_line if isinstance(finding.span, SourceSpan) else 0
place = (path, line, str(finding.metadata.get("symbol")))
levels[place] = max(level, levels.get(place, level), key=REACHED.index)
return "; ".join(
f"{path}:{line} {symbol} ({level})" for (path, line, symbol), level in sorted(levels.items())
f"{path}{f':{line}' if line else ''} {symbol} ({level})"
for (path, line, symbol), level in sorted(levels.items())
)


Expand Down
Loading
Loading