diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d1f88c..912bea8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/docs/usage.md b/docs/usage.md index db7b66e..62815ca 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -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"} ] } @@ -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: diff --git a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py index f66a402..96b83b3 100644 --- a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py +++ b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -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 @@ -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", @@ -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 diff --git a/src/coretrace_python/bundled/dependency/vulnerable_dependency/plugin.toml b/src/coretrace_python/bundled/dependency/vulnerable_dependency/plugin.toml index 1e94a53..269c4d2 100644 --- a/src/coretrace_python/bundled/dependency/vulnerable_dependency/plugin.toml +++ b/src/coretrace_python/bundled/dependency/vulnerable_dependency/plugin.toml @@ -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] diff --git a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py index 8c77036..edf40a2 100644 --- a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py +++ b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py @@ -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", @@ -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 diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 68ecbe2..dd3aa91 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -22,6 +22,7 @@ from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.interprocedural import ( + Arguments, CallSite, ExternalCall, ExternalSymbol, @@ -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) @@ -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), } @@ -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"]), ) @@ -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), } @@ -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, ) diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index 676558f..99b8c9f 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -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 @@ -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, ) diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index ad2dbd5..6dff7d2 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -10,10 +10,18 @@ from __future__ import annotations from collections.abc import Iterable, Mapping - -from coretrace_python.dependency.graph import DIRECT, Advisory, DependencyGraph +from dataclasses import dataclass + +from coretrace_python.dependency.graph import ( + DIRECT, + Advisory, + AdvisoryEntryPoint, + Condition, + DependencyGraph, +) from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.findings.refutation import Status, Verdict, Verdicts +from coretrace_python.interprocedural import Arguments, CallSite, ExternalSymbol from coretrace_python.semantic.symbols import SymbolId from coretrace_python.taint import Sink, TaintFlow, TaintKind @@ -40,10 +48,68 @@ def advisory_sinks(affected: Affected) -> tuple[Sink, ...]: return tuple(Sink(symbol, TaintKind.ADVISORY) for symbol in affected) -def evidence(advisory: Advisory, symbol: SymbolId, level: str) -> dict[str, str]: +@dataclass(frozen=True) +class ConditionCheck: + """What one call tells of an entry point's conditions: those it meets, those left to + review (every semantic one, and every argument one the call does not decide), and + the argument condition it contradicts with what it passes instead (None: absent), + in which case the call does not reach the vulnerability.""" + + met: tuple[Condition, ...] = () + pending: tuple[Condition, ...] = () + contradicted: Condition | None = None + passed: str | None = None + + +def check_conditions(entry: AdvisoryEntryPoint | None, arguments: Arguments | None) -> ConditionCheck: + """Decide ``entry``'s conditions against what the call's ``arguments`` denote.""" + + if entry is None: + return ConditionCheck() + met: list[Condition] = [] + pending: list[Condition] = [] + for condition in entry.conditions: + given = _given(condition, arguments) if condition.checkable and arguments is not None else None + if given is None: + pending.append(condition) + continue + explicit, value = given + if explicit and value is None: + pending.append(condition) + continue + affected = value in condition.values if explicit else condition.default + if not affected: + return ConditionCheck(tuple(met), tuple(pending), condition, value) + met.append(condition) + return ConditionCheck(tuple(met), tuple(pending)) + + +def _given(condition: Condition, arguments: Arguments) -> tuple[bool, str | None] | None: + """Whether the call gives the condition's argument, and what it denotes: ``(True, + value)`` when given, ``(False, None)`` when surely absent, None when it cannot tell.""" + + for name, value in arguments.keywords: + if name == condition.argument: + return True, value + if condition.position is not None and condition.position < len(arguments.positional): + return True, arguments.positional[condition.position] + return None if arguments.unpacked else (False, None) + + +def ruled_out(module: str, site: CallSite, check: ConditionCheck) -> str: + """Which call a contradicted condition rules out, and why: ``app:12 + python.yaml.load(Loader=python.yaml.SafeLoader)``.""" + + assert check.contradicted is not None and isinstance(site.target, ExternalSymbol) + argument = check.contradicted.argument or f"#{check.contradicted.position}" + passed = f"{argument} absent" if check.passed is None else f"{argument}={check.passed}" + return f"{module}:{site.location.start_line} {site.target.symbol}({passed})" + + +def evidence(advisory: Advisory, symbol: SymbolId, level: str, check: ConditionCheck) -> dict[str, str]: """What a finding keeps of the advisory for ``symbol``: the level of evidence - established, how the symbol relates to the vulnerability and under which - conditions, the ones the engine could not check listed as pending review.""" + established, how the symbol relates to the vulnerability, and its conditions — those + the call meets and those left to review.""" metadata = {"advisory": advisory.id, "package": advisory.package, "symbol": str(symbol), "level": level} entry = advisory.entry_point(symbol) @@ -54,9 +120,10 @@ def evidence(advisory: Advisory, symbol: SymbolId, level: str) -> dict[str, str] metadata["justification"] = entry.justification if entry.conditions: metadata["conditions"] = "; ".join(c.text for c in entry.conditions) - # ponytail: no condition is checked yet, so every one awaits review; argument - # conditions get checked at the call site once call sites carry argument symbols. - metadata["conditions_pending_review"] = "; ".join(c.text for c in entry.conditions) + if check.met: + metadata["conditions_met"] = "; ".join(c.text for c in check.met) + if check.pending: + metadata["conditions_pending_review"] = "; ".join(c.text for c in check.pending) return metadata @@ -77,21 +144,22 @@ def correlate( if verdict is not None and verdict.status is Status.REFUTED: continue hotspot = verdict is not None and verdict.status is Status.HOTSPOT - findings.extend( - _exploitable(function, flow, advisory, verdict, hotspot) for advisory in affected.get(flow.sink.symbol, ()) - ) + for advisory in affected.get(flow.sink.symbol, ()): + check = check_conditions(advisory.entry_point(flow.sink.symbol), flow.sink_arguments) + if check.contradicted is None: + findings.append(_exploitable(function, flow, advisory, verdict, hotspot, check)) return tuple(findings) def _exploitable( - function: str, flow: TaintFlow, advisory: Advisory, verdict: Verdict | None, hotspot: bool + function: str, flow: TaintFlow, advisory: Advisory, verdict: Verdict | None, hotspot: bool, check: ConditionCheck ) -> Finding: message = ( f"{advisory.id}: {flow.source.label} input reaches {flow.sink.symbol}, affected in " f"the required {advisory.package} {advisory.vulnerable}: {advisory.summary}" ) metadata = { - **evidence(advisory, flow.sink.symbol, "exploitable"), + **evidence(advisory, flow.sink.symbol, "exploitable", check), "source": str(flow.source.symbol), "source_label": flow.source.label, "verdict": "hotspot" if hotspot else "vulnerability", diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 52f97fc..3450ec2 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -147,13 +147,18 @@ class Condition: """What must hold for an entry point to reach, or exploit, the affected code: the value of an argument, the loader or algorithm used, the format decoded, the kind of input, a configuration. ``kind`` says whether the engine can check it: an - ``argument`` condition names the argument and the values that satisfy it; a - ``semantic`` condition cannot be checked yet and is reported as pending review.""" + ``argument`` condition names the argument by keyword, and by ``position`` when it may + be passed positionally, and lists the ``values`` that satisfy it — symbols + (``python.yaml.FullLoader``) or constants as Python writes them (``True``); + ``default`` says an absent argument means a vulnerable value. A ``semantic`` + condition cannot be checked and is reported as pending review.""" kind: str text: str argument: str | None = None values: tuple[str, ...] = () + position: int | None = None + default: bool = False @property def checkable(self) -> bool: diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index 379d9fb..e2ed9cd 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -1,6 +1,7 @@ """Call graph and function summaries (architecture §19, §20).""" from coretrace_python.interprocedural.callgraph import ( + Arguments, CallGraph, CallGraphAnalysis, CallSite, @@ -29,6 +30,7 @@ ) __all__ = [ + "Arguments", "CallGraph", "CallGraphAnalysis", "CallSite", diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index f589756..85c581b 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -9,7 +9,7 @@ from __future__ import annotations from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from types import MappingProxyType from typing import ClassVar @@ -25,10 +25,12 @@ from coretrace_python.ir.model import ( Await, Call, + Constant, FunctionIR, GetAttr, GetItem, Global, + Instruction, MakeFunction, Symbol, Value, @@ -58,13 +60,25 @@ class UnknownTarget: Target = KnownFunction | ExternalSymbol | UnknownTarget +@dataclass(frozen=True) +class Arguments: + """What the arguments of a call denote: a symbol's canonical name + (``python.yaml.FullLoader``), a constant as Python writes it (``True``, ``'/static'``), + or None for anything else (a parameter, an expression). ``unpacked`` says the call + unpacks ``*args`` or ``**kwargs``, so an argument it does not give explicitly may + still be given; after ``*args`` no position is known, and ``positional`` is empty.""" + + positional: tuple[str | None, ...] = () + keywords: tuple[tuple[str, str | None], ...] = () + unpacked: bool = False + + @dataclass(frozen=True) class CallSite: caller: str location: SourceSpan target: Target - arguments: int - keywords: int + arguments: Arguments = field(default_factory=Arguments) @dataclass(frozen=True) @@ -93,9 +107,7 @@ def __init__( self._names = {function.span: name for name, function in definitions.items()} self.unsupported = unsupported self._sites = MappingProxyType(dict(sites)) - self._targets = { - (site.caller, site.location): site.target for found in sites.values() for site in found - } + self._at = {(site.caller, site.location): site for found in sites.values() for site in found} callers: dict[str, set[str]] = {name: set() for name in definitions} for found in sites.values(): for site in found: @@ -115,7 +127,14 @@ def sites(self, caller: str) -> tuple[CallSite, ...]: return self._sites.get(caller, ()) def target_at(self, caller: str, location: SourceSpan) -> Target: - return self._targets.get((caller, location), UnknownTarget()) + site = self._at.get((caller, location)) + return site.target if site is not None else UnknownTarget() + + def arguments_at(self, caller: str, location: SourceSpan) -> Arguments: + """What the arguments of the call at ``location`` in ``caller`` denote.""" + + site = self._at.get((caller, location)) + return site.arguments if site is not None else Arguments() def callees(self, caller: str) -> frozenset[str]: return frozenset( @@ -165,6 +184,29 @@ def derive_symbols( return symbols +# Conditions name short constants (``True``, ``None``, a mode); a log message or a query +# passed as an argument is not recorded, so call sites stay small in the cache. +MAX_CONSTANT_LENGTH = 64 + + +def _arguments(call: Call, symbols: Mapping[Value, SymbolId], defs: Mapping[Value, Instruction]) -> Arguments: + def denoted(value: Value) -> str | None: + symbol = symbols.get(value) + if symbol is not None: + return str(symbol) + made = defs.get(value) + if not isinstance(made, Constant): + return None + written = repr(made.value) + return written if len(written) <= MAX_CONSTANT_LENGTH else None + + return Arguments( + () if call.starred else tuple(denoted(value) for value in call.arguments), + tuple((name, denoted(value)) for name, value in call.keywords if name is not None), + bool(call.starred) or any(name is None for name, _ in call.keywords), + ) + + def resolve_targets( function: FunctionIR, scopes: ScopeTable, @@ -339,6 +381,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: _typed_with_module_classes(function, ssa, classes), bases, ) + defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None} found: list[CallSite] = [] for block in ssa.blocks: for instruction in block.instructions: @@ -348,8 +391,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: name, instruction.location, targets.get(instruction.callee, UnknownTarget()), - len(instruction.arguments), - len(instruction.keywords), + _arguments(instruction, symbols[name], defs), ) ) sites[name] = tuple(found) diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index a16e4df..030af3f 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -9,6 +9,7 @@ from __future__ import annotations +import dataclasses from collections.abc import Mapping from dataclasses import dataclass from types import MappingProxyType @@ -26,6 +27,7 @@ from coretrace_python.cfg import CFG, BlockId, CFGAnalysis from coretrace_python.dataflow import DataflowProblem, Direction, solve from coretrace_python.interprocedural.callgraph import ( + Arguments, CallGraph, CallGraphAnalysis, ExternalSymbol, @@ -92,6 +94,8 @@ class ExternalCall: keyword_dependencies: Dependencies location: SourceSpan call_site: SourceSpan | None + # What the arguments of the call at ``location`` denote, where it is made. + arguments: Arguments = dataclasses.field(default_factory=Arguments) @dataclass(frozen=True) @@ -334,6 +338,7 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: keywords.parameters, call.location, None, + self.graph.arguments_at(self.name, call.location), ) return everything | Dep(externals=frozenset({target.symbol})) if isinstance(target, KnownFunction): @@ -416,6 +421,7 @@ def mapped(indices: Dependencies) -> Dep: mapped(reached.keyword_dependencies).parameters, reached.location, call.location, + reached.arguments, ) made = self.defs.get(call.callee) values = (*receiver, *call.arguments, *(made.captured if isinstance(made, MakeFunction) else ())) @@ -432,6 +438,7 @@ def record( keywords: Dependencies, location: SourceSpan, call_site: SourceSpan | None, + given: Arguments, ) -> None: key = (symbol, location, call_site) previous = self.external.get(key) @@ -440,7 +447,7 @@ def record( a | b for a, b in zip(previous.argument_dependencies, arguments, strict=False) ) keywords |= previous.keyword_dependencies - self.external[key] = ExternalCall(symbol, arguments, keywords, location, call_site) + self.external[key] = ExternalCall(symbol, arguments, keywords, location, call_site, given) def summarize( diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 30d657e..c1efe21 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -27,6 +27,7 @@ from coretrace_python.dataflow import DataflowProblem, Direction, solve from coretrace_python.hir import nodes from coretrace_python.interprocedural import ( + Arguments, CallGraph, CallGraphAnalysis, ExternalSymbol, @@ -112,6 +113,8 @@ class TaintFlow: location: SourceSpan through: str | None = None sink_location: SourceSpan | None = None + # What the arguments of the sink call denote, to decide the conditions on them. + sink_arguments: Arguments | None = None Key = Value | HeapLocation @@ -395,10 +398,11 @@ def external( sink = self.models.sink(symbol) if sink is not None: + given = self.graph.arguments_at(self.name, call.location) for position, argument in enumerate(call.arguments): - self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, position) + self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, position, given) for argument in (*call.starred, *(value for _, value in call.keywords)): - self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, None) + self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, None, given) sanitizer = self.models.sanitizer(symbol) if sanitizer is not None: return everything.without(sanitizer.kinds) @@ -484,7 +488,7 @@ def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: for position, deps in zip(positions, (*reached.argument_dependencies, reached.keyword_dependencies), strict=True): taint, witness = mapped(deps) if witness is not None: - self.report(flows, sink, taint, witness, call, through, reached.location, position) + self.report(flows, sink, taint, witness, call, through, reached.location, position, reached.arguments) for mutation in summary.mutations: if mutation.parameter < len(values): stored = mapped(mutation.dependencies)[0] @@ -510,6 +514,7 @@ def report( through: str | None, sink_location: SourceSpan | None, position: int | None = None, + arguments: Arguments | None = None, ) -> None: reaching = taint.kinds & sink.kinds_at(position) if not reaching: @@ -524,6 +529,7 @@ def report( call.location, through, call.location if sink_location is None else sink_location, + arguments, ) ) diff --git a/tests/test_advisory_evidence.py b/tests/test_advisory_evidence.py index 3f7fe14..b80390e 100644 --- a/tests/test_advisory_evidence.py +++ b/tests/test_advisory_evidence.py @@ -41,7 +41,14 @@ SymbolId("python.vulnlib.parse"), "calls _parser.parse_unsafe when mode is UNSAFE (commit 1234abc)", ( - Condition("argument", "mode is UNSAFE, the default", argument="mode", values=("python.vulnlib.UNSAFE",)), + Condition( + "argument", + "mode is UNSAFE, the default", + argument="mode", + values=("python.vulnlib.UNSAFE",), + position=1, + default=True, + ), Condition("semantic", "the document carries a custom tag"), ), ), diff --git a/tests/test_argument_conditions.py b/tests/test_argument_conditions.py new file mode 100644 index 0000000..75da427 --- /dev/null +++ b/tests/test_argument_conditions.py @@ -0,0 +1,217 @@ +"""Argument conditions are decided at the call site. + +An advisory entry point may be affected for some values of one argument only: +``yaml.load`` with ``Loader=FullLoader`` but not with ``SafeLoader``, ``add_static`` +with ``follow_symlinks=True`` only. Each call records what its arguments denote — a +symbol, or a constant as Python writes it — so a condition is met, contradicted (the +call does not reach the vulnerability, and the vulnerable requirement says which call +was ruled out and why) or left pending review when the call does not tell. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + AdvisoryFileError, + Condition, + dump_advisories, + load_advisories, +) +from coretrace_python.findings import Finding, Severity +from coretrace_python.semantic.symbols import SymbolId + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "src" / "coretrace_python" / "bundled" + +MODE = Condition( + "argument", + "mode is UNSAFE, the default", + argument="mode", + values=("python.vulnlib.UNSAFE",), + position=1, + default=True, +) +TAG = Condition("semantic", "the document carries a custom tag") +FOLLOW = Condition("argument", "follow is True", argument="follow", values=("True",)) + +PARSE = Advisory( + "CVE-2099-0101", + "vulnlib", + "<1.1", + "parse runs code from the document in unsafe mode", + Severity.CRITICAL, + entry_points=(AdvisoryEntryPoint(SymbolId("python.vulnlib.parse"), "commit 1234abc", (MODE, TAG)),), + modules=("vulnlib",), +) +SERVE = Advisory( + "CVE-2099-0102", + "vulnlib", + "<1.1", + "serve follows links out of its root", + Severity.HIGH, + entry_points=(AdvisoryEntryPoint(SymbolId("python.vulnlib.serve"), "commit 5678def", (FOLLOW,)),), + modules=("vulnlib",), +) + + +def analyse(root: Path, files: dict[str, str], **options: object) -> engine.ProjectAnalysis: + for relative, text in {"requirements.txt": "vulnlib==1.0\n", **files}.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((PARSE, SERVE)), encoding="utf-8") + return engine.analyze_project(root, [PLUGINS], **options) # type: ignore[arg-type] + + +def calling(call: str) -> dict[str, str]: + return {"app.py": f"import vulnlib\n\ndef run(text, path, choice, options, rest):\n return {call}\n"} + + +def of_rule(findings: tuple[Finding, ...], rule_id: str) -> list[Finding]: + return [f for f in findings if f.rule_id == rule_id] + + +# --------------------------------------------------------------------------- met + + +@pytest.mark.parametrize( + "call", + ["vulnlib.parse(text, mode=vulnlib.UNSAFE)", "vulnlib.parse(text, vulnlib.UNSAFE)", "vulnlib.parse(text)"], +) +def test_a_call_that_meets_the_argument_condition_is_reachable_with_the_condition_met(tmp_path: Path, call: str) -> None: + (found,) = of_rule(analyse(tmp_path, calling(call)).findings, "reachable-vulnerability") + + assert found.metadata["conditions_met"] == "mode is UNSAFE, the default" + assert found.metadata["conditions_pending_review"] == "the document carries a custom tag" + + +def test_a_constant_argument_meets_its_condition_as_python_writes_it(tmp_path: Path) -> None: + (found,) = of_rule(analyse(tmp_path, calling("vulnlib.serve(path, follow=True)")).findings, "reachable-vulnerability") + + assert found.metadata["conditions_met"] == "follow is True" + assert "conditions_pending_review" not in found.metadata + + +# --------------------------------------------------------------------------- contradicted + + +@pytest.mark.parametrize( + "call, ruled_out", + [ + ("vulnlib.parse(text, mode=vulnlib.SAFE)", "app:4 python.vulnlib.parse(mode=python.vulnlib.SAFE)"), + ("vulnlib.parse(text, vulnlib.SAFE)", "app:4 python.vulnlib.parse(mode=python.vulnlib.SAFE)"), + ("vulnlib.parse(text, mode=None)", "app:4 python.vulnlib.parse(mode=None)"), + ("vulnlib.serve(path, follow=False)", "app:4 python.vulnlib.serve(follow=False)"), + ("vulnlib.serve(path)", "app:4 python.vulnlib.serve(follow absent)"), + ], +) +def test_a_call_that_contradicts_an_argument_condition_is_ruled_out_with_its_reason( + tmp_path: Path, call: str, ruled_out: str +) -> None: + findings = analyse(tmp_path, calling(call)).findings + + assert of_rule(findings, "reachable-vulnerability") == [] + dependencies = of_rule(findings, "vulnerable-dependency") + assert [f.metadata.get("ruled_out") for f in dependencies if "ruled_out" in f.metadata] == [ruled_out] + assert {f.metadata["level"] for f in dependencies} == {"imported"} + + +# --------------------------------------------------------------------------- undecided + + +@pytest.mark.parametrize( + "call", ["vulnlib.parse(text, mode=choice)", "vulnlib.parse(text, **options)", "vulnlib.parse(*rest)"] +) +def test_a_call_that_does_not_tell_leaves_the_condition_pending_review(tmp_path: Path, call: str) -> None: + (found,) = of_rule(analyse(tmp_path, calling(call)).findings, "reachable-vulnerability") + + assert "conditions_met" not in found.metadata + assert found.metadata["conditions_pending_review"] == "mode is UNSAFE, the default; the document carries a custom tag" + + +# --------------------------------------------------------------------------- exploitable + + +def test_attacker_input_is_exploitable_only_through_a_call_that_meets_the_conditions(tmp_path: Path) -> None: + source = "import vulnlib\n\ndef run():\n vulnlib.parse(input(), mode=vulnlib.SAFE)\n vulnlib.parse(input())\n" + + exploitable = of_rule(analyse(tmp_path, {"app.py": source}).findings, "exploitable-vulnerability") + + assert [(f.span.start_line, f.metadata["conditions_met"]) for f in exploitable] == [(5, "mode is UNSAFE, the default")] + + +def test_the_sink_call_decides_when_the_flow_goes_through_a_function_of_another_module(tmp_path: Path) -> None: + files = { + "app/__init__.py": "", + "app/loader.py": ( + "import vulnlib\n\n" + "def safe(text):\n return vulnlib.parse(text, mode=vulnlib.SAFE)\n\n" + "def unsafe(text):\n return vulnlib.parse(text)\n" + ), + "app/main.py": "from app.loader import safe, unsafe\n\ndef run():\n safe(input())\n unsafe(input())\n", + } + + exploitable = of_rule(analyse(tmp_path, files).findings, "exploitable-vulnerability") + + assert [(Path(str(f.span.source_id)).name, f.span.start_line, f.metadata.get("through")) for f in exploitable] == [ + ("main.py", 5, "app.loader.unsafe") + ] + + +# --------------------------------------------------------------------------- cache and files + + +def test_a_module_served_from_the_cache_decides_the_same(tmp_path: Path) -> None: + files = {"app.py": "import vulnlib\n\ndef run(text, path):\n vulnlib.parse(text)\n vulnlib.serve(path)\n"} + cache = ProjectCache(tmp_path / "cache") + + first = analyse(tmp_path / "src", files, cache=cache) + second = analyse(tmp_path / "src", files, cache=cache) + + assert "app" in second.reused + assert second.findings == first.findings + assert [f.metadata["advisory"] for f in of_rule(second.findings, "reachable-vulnerability")] == ["CVE-2099-0101"] + assert "app:5 python.vulnlib.serve(follow absent)" in [f.metadata.get("ruled_out") for f in second.findings] + + +def test_argument_conditions_round_trip_through_advisory_files(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((PARSE, SERVE)), encoding="utf-8") + + assert load_advisories(path) == (PARSE, SERVE) + conditions = [e["conditions"][0] for a in json.loads(path.read_text())["advisories"] for e in a["entry_points"]] + assert conditions[0] == { + "kind": "argument", + "text": "mode is UNSAFE, the default", + "argument": "mode", + "values": ["python.vulnlib.UNSAFE"], + "position": 1, + "default": True, + } + assert "position" not in conditions[1] and "default" not in conditions[1] + + +@pytest.mark.parametrize("field, value", [("position", "1"), ("default", "yes")]) +def test_a_malformed_argument_condition_is_rejected(tmp_path: Path, field: str, value: str) -> None: + path = tmp_path / "advisories.json" + document = json.loads(dump_advisories((PARSE,))) + document["advisories"][0]["entry_points"][0]["conditions"][0][field] = value + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError, match=field): + load_advisories(path) + + +def test_a_long_constant_is_not_recorded_and_leaves_the_condition_pending(tmp_path: Path) -> None: + long = "x" * 100 + (found,) = of_rule(analyse(tmp_path, calling(f"vulnlib.serve(path, follow='{long}')")).findings, "reachable-vulnerability") + + assert found.metadata["conditions_pending_review"] == "follow is True" diff --git a/tests/test_cache.py b/tests/test_cache.py index 419a7ae..5b5fafa 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -68,27 +68,37 @@ def rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: def test_findings_and_summaries_round_trip_through_json() -> None: - from coretrace_python.interprocedural import ExternalCall, FunctionSummary, ModuleFunction + from coretrace_python.interprocedural import ( + Arguments, + CallSite, + ExternalCall, + ExternalSymbol, + FunctionSummary, + ModuleFunction, + ) from coretrace_python.semantic.symbols import SymbolId span = SourceSpan(SourceId("/p/a.py"), 3, 5, 3, 9) finding = Finding("r", "m", Severity.HIGH, Confidence.MEDIUM, span, "f", {"k": "v"}) + arguments = Arguments(("python.yaml.Loader", None, "True"), (("mode", "'x'"), ("other", None)), True) summary = FunctionSummary( "f", 2, frozenset({0}), - (ExternalCall(SymbolId("python.os.system"), (frozenset({0}), frozenset()), frozenset({1}), span, None),), + (ExternalCall(SymbolId("python.os.system"), (frozenset({0}), frozenset()), frozenset({1}), span, None, arguments),), False, frozenset({SymbolId("python.builtins.input")}), ) functions = (ModuleFunction("f", span, "http"), ModuleFunction("g", span)) - text = json.dumps(encode(CachedModule(functions, {"f": summary}, (), (finding,)))) + site = CallSite("f", span, ExternalSymbol(SymbolId("python.yaml.load")), arguments) + text = json.dumps(encode(CachedModule(functions, {"f": summary}, (site,), (finding,)))) restored = decode(json.loads(text)) assert restored.findings == (finding,) assert restored.summaries["f"] == summary assert restored.functions == functions + assert restored.sites == (site,) # --------------------------------------------------------------------------- keys diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py index fe70a29..3b6b93d 100644 --- a/tests/test_dependencies.py +++ b/tests/test_dependencies.py @@ -281,7 +281,7 @@ def test_sample_advisories_are_shipped() -> None: 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["vulnerable-dependency"].manifest.requires == ("dependency.graph", "interprocedural.callgraph") assert loaded["reachable-vulnerability"].manifest.requires == ("dependency.graph", "interprocedural.callgraph")