Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,11 @@
### Plugins

- `ProjectContext.functions(module)` gives project plugins every function of a module as its call graph names it, with its span and the label of the entry point it is (`http` for a route, `argv` for a command), decided by the taint engine's own rule; modules served from the cache keep them, and the cache format is bumped (#116).
- Call sites and external calls record what each argument denotes — a symbol, a constant as Python writes it, or nothing known — in an `Arguments` record: `CallSite.arguments` (formerly the argument counts), `ExternalCall.arguments`, and `TaintFlow.sink_arguments` for the sink call of a flow, through callees in other modules too (#110).

### Dependencies

- Argument conditions of advisory entry points are decided at each call, with a `position` for positional arguments and a `default` for absent ones: a call meeting them is reported with `conditions_met`, a call passing another known value does not reach the vulnerability and is listed in the requirement's `ruled_out`, and a call the analyzer cannot decide leaves them pending review (#110).

## 0.4.0 (2026-09-24)

Expand Down
32 changes: 24 additions & 8 deletions docs/usage.md
Original file line number Diff line number Diff line change
Expand Up @@ -271,7 +271,8 @@ affected APIs that feed reachability and correlation:
"symbol": "python.yaml.load",
"justification": "same path when Loader is FullLoader",
"conditions": [
{"kind": "argument", "text": "Loader is FullLoader", "argument": "Loader", "values": ["python.yaml.FullLoader"]},
{"kind": "argument", "text": "Loader is FullLoader, the default", "argument": "Loader", "position": 1,
"values": ["python.yaml.FullLoader", "python.yaml.loader.FullLoader"], "default": true},
{"kind": "semantic", "text": "the document carries a python/object/apply tag"}
]
}
Expand All @@ -285,17 +286,32 @@ affected APIs that feed reachability and correlation:

`affected_symbols` are the functions the fix changed; `entry_points` the public APIs
through which a project reaches them, each with the justification that ties it to the
fix — the commit, or the call path — and the `conditions` under which it is affected: an
argument value the engine can check, or a `semantic` condition it cannot yet, kept on the
finding as pending review rather than dropped. A call to either kind of symbol is
reachable. `modules` names the top-level modules the package installs, so the analyzer
can tell an imported package from a merely required one; it defaults to the package name.
fix — the commit, or the call path — and the `conditions` under which it is affected. A
call to either kind of symbol is reachable. `modules` names the top-level modules the
package installs, so the analyzer can tell an imported package from a merely required
one; it defaults to the package name.

An `argument` condition is decided at each call. It names the argument by keyword and,
when it may be passed positionally, by `position` (counted from 0, the receiver of a
method excluded); `values` are what makes the call affected, written as the analyzer
sees arguments: a symbol by its canonical name (`python.yaml.FullLoader`, and every
spelling a project may use, such as `python.yaml.loader.FullLoader`), a constant as
Python writes it (`True`, `None`, `'/static'`); `default` says an absent argument means
an affected value. A call passing another known value does not reach the vulnerability;
a call passing a variable, or unpacking `*args` or `**kwargs`, leaves the condition
pending review. A `semantic` condition — the document's tags, the platform, the
template's origin — cannot be decided and is always pending review, never dropped.

Every dependency finding records the highest level of evidence established in its
`level` metadata: `declared` (the requirement allows a vulnerable version), `imported`
(a module of the package is imported somewhere), `reachable` (an entry point or affected
symbol is called) or `exploitable` (attacker input reaches it), with `entry_point`,
`justification`, `conditions` and `conditions_pending_review` on the last two.
symbol is called) or `exploitable` (attacker input reaches it). The last two carry
`entry_point`, `justification`, `conditions`, and which of them the call meets
(`conditions_met`) or leaves to review (`conditions_pending_review`). A requirement whose
entry points are only called with arguments that contradict a condition stays
`imported`, and its finding lists those calls in `ruled_out`
(`app.views:12 python.yaml.load(Loader=python.yaml.SafeLoader)`), the evidence that
they do not reach the vulnerability.

A `coretrace-policy.toml` at the root, or the file passed with `--policy`, denies
packages, requires pins and lists accepted advisories whose findings are dropped:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

from coretrace_python.analysis import AnyAnalysis
from coretrace_python.dependency import DependencyAnalysis
from coretrace_python.dependency.correlation import affected_symbols, evidence
from coretrace_python.dependency.correlation import affected_symbols, check_conditions, evidence
from coretrace_python.findings import Confidence, Finding
from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol
from coretrace_python.plugins import ProjectContext, ProjectPlugin
Expand All @@ -29,6 +29,9 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
if not isinstance(site.target, ExternalSymbol):
continue
for advisory in affected.get(site.target.symbol, ()):
check = check_conditions(advisory.entry_point(site.target.symbol), site.arguments)
if check.contradicted is not None:
continue
findings.append(
Finding(
rule_id="reachable-vulnerability",
Expand All @@ -40,7 +43,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
confidence=Confidence.HIGH,
span=site.location,
function=function,
metadata=evidence(advisory, site.target.symbol, "reachable"),
metadata=evidence(advisory, site.target.symbol, "reachable", check),
)
)
return findings
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
name = "vulnerable-dependency"
version = "1.0.0"
plugin_api = ">=1,<2"
requires = ["dependency.graph"]
requires = ["dependency.graph", "interprocedural.callgraph"]
provides = ["vulnerability.vulnerable-dependency"]

[entrypoint]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,23 +7,34 @@

from coretrace_python.analysis import AnyAnalysis
from coretrace_python.dependency import DependencyAnalysis
from coretrace_python.dependency.correlation import affected_symbols, check_conditions, ruled_out
from coretrace_python.findings import Confidence, Finding
from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol
from coretrace_python.plugins import ProjectContext, ProjectPlugin


class VulnerableDependencyPlugin(ProjectPlugin):
name: ClassVar[str] = "vulnerable-dependency"
requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({DependencyAnalysis})
requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({DependencyAnalysis, CallGraphAnalysis})

def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
findings: list[Finding] = []
imported = [s for module in ctx.modules for s in ctx.imports(module).all_symbols()]
excluded = _ruled_out(ctx)
for requirement in ctx.dependencies.requirements:
for advisory in ctx.advisories:
if not advisory.affects(requirement):
continue
pinned = requirement.pinned is not None
level = "imported" if advisory.imported_by(imported) else "declared"
metadata = {
"advisory": advisory.id,
"package": advisory.package,
"specifier": requirement.specifier,
"level": level,
}
if excluded.get(advisory.id):
metadata["ruled_out"] = "; ".join(excluded[advisory.id])
findings.append(
Finding(
rule_id="vulnerable-dependency",
Expand All @@ -35,12 +46,26 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
severity=advisory.severity,
confidence=Confidence.HIGH if pinned else Confidence.MEDIUM,
span=requirement.span,
metadata={
"advisory": advisory.id,
"package": advisory.package,
"specifier": requirement.specifier,
"level": level,
},
metadata=metadata,
)
)
return findings


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."""

affected = affected_symbols(ctx.dependencies, ctx.advisories)
excluded: dict[str, list[str]] = {}
for module in sorted(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
for advisory in affected.get(site.target.symbol, ()):
check = check_conditions(advisory.entry_point(site.target.symbol), site.arguments)
if check.contradicted is not None:
excluded.setdefault(advisory.id, []).append(ruled_out(module, site, check))
return excluded
33 changes: 28 additions & 5 deletions src/coretrace_python/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

from coretrace_python.findings import Confidence, Finding, Severity
from coretrace_python.interprocedural import (
Arguments,
CallSite,
ExternalCall,
ExternalSymbol,
Expand All @@ -38,7 +39,7 @@
from coretrace_python.semantic.symbols import SymbolId
from coretrace_python.source import SourceId, SourceSpan

CACHE_FORMAT = 6
CACHE_FORMAT = 7


@dataclass(frozen=True)
Expand Down Expand Up @@ -225,6 +226,7 @@ def _encode_call(call: ExternalCall) -> dict[str, Any]:
"keywords": sorted(call.keyword_dependencies),
"location": _encode_span(call.location),
"call_site": None if call.call_site is None else _encode_span(call.call_site),
"given": _encode_arguments(call.arguments),
}


Expand All @@ -236,6 +238,7 @@ def _decode_call(data: Mapping[str, Any]) -> ExternalCall:
_indices(data["keywords"]),
_decode_span(data["location"]),
None if site is None else _decode_span(site),
_decode_arguments(data["given"]),
)


Expand Down Expand Up @@ -317,8 +320,7 @@ def _encode_site(site: CallSite) -> dict[str, Any]:
"caller": site.caller,
"location": _encode_span(site.location),
"target": _encode_target(site.target),
"arguments": site.arguments,
"keywords": site.keywords,
"arguments": _encode_arguments(site.arguments),
}


Expand All @@ -327,6 +329,27 @@ def _decode_site(data: Mapping[str, Any]) -> CallSite:
_string(data["caller"]),
_decode_span(data["location"]),
_decode_target(data["target"]),
_integer(data["arguments"]),
_integer(data["keywords"]),
_decode_arguments(data["arguments"]),
)


def _encode_arguments(arguments: Arguments) -> dict[str, Any]:
return {
"positional": list(arguments.positional),
"keywords": [[name, value] for name, value in arguments.keywords],
"unpacked": arguments.unpacked,
}


def _decode_arguments(data: Mapping[str, Any]) -> Arguments:
def denoted(value: Any) -> str | None:
return None if value is None else _string(value)

unpacked = data["unpacked"]
if not isinstance(unpacked, bool):
raise TypeError(f"expected a boolean, got {unpacked!r}")
return Arguments(
tuple(denoted(value) for value in data["positional"]),
tuple((_string(name), denoted(value)) for name, value in data["keywords"]),
unpacked,
)
31 changes: 22 additions & 9 deletions src/coretrace_python/dependency/advisories.py
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,10 @@ def _condition_entry(condition: Condition) -> dict[str, Any]:
entry["argument"] = condition.argument
if condition.values:
entry["values"] = list(condition.values)
if condition.position is not None:
entry["position"] = condition.position
if condition.default:
entry["default"] = True
return entry


Expand Down Expand Up @@ -193,13 +197,22 @@ def _entry_point(entry: Mapping[str, Any]) -> AdvisoryEntryPoint:
return AdvisoryEntryPoint(
SymbolId(str(entry["symbol"])),
str(entry["justification"]),
tuple(
Condition(
str(c["kind"]),
str(c["text"]),
None if c.get("argument") is None else str(c["argument"]),
tuple(str(v) for v in c.get("values") or []),
)
for c in entry.get("conditions") or []
),
tuple(_condition(c) for c in entry.get("conditions") or []),
)


def _condition(entry: Mapping[str, Any]) -> Condition:
position = entry.get("position")
if position is not None and (not isinstance(position, int) or isinstance(position, bool) or position < 0):
raise TypeError(f"condition position must be a non-negative integer, got {position!r}")
default = entry.get("default", False)
if not isinstance(default, bool):
raise TypeError(f"condition default must be true or false, got {default!r}")
return Condition(
str(entry["kind"]),
str(entry["text"]),
None if entry.get("argument") is None else str(entry["argument"]),
tuple(str(v) for v in entry.get("values") or []),
position,
default,
)
Loading
Loading