From b304bd5db40f56e16009c24576bb45c04d95f9ae Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 24 Sep 2026 17:53:36 +0200 Subject: [PATCH 1/2] test: specify reachability through reads of advisory entry points --- tests/test_cache.py | 5 +- tests/test_read_entry_points.py | 123 ++++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 1 deletion(-) create mode 100644 tests/test_read_entry_points.py diff --git a/tests/test_cache.py b/tests/test_cache.py index 5b5fafa..ad9db2d 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -75,6 +75,7 @@ def test_findings_and_summaries_round_trip_through_json() -> None: ExternalSymbol, FunctionSummary, ModuleFunction, + SymbolRead, ) from coretrace_python.semantic.symbols import SymbolId @@ -92,13 +93,15 @@ def test_findings_and_summaries_round_trip_through_json() -> None: functions = (ModuleFunction("f", span, "http"), ModuleFunction("g", span)) site = CallSite("f", span, ExternalSymbol(SymbolId("python.yaml.load")), arguments) - text = json.dumps(encode(CachedModule(functions, {"f": summary}, (site,), (finding,)))) + read = SymbolRead("f", span, SymbolId("python.flask.request.form")) + text = json.dumps(encode(CachedModule(functions, {"f": summary}, (site,), (finding,), (read,)))) restored = decode(json.loads(text)) assert restored.findings == (finding,) assert restored.summaries["f"] == summary assert restored.functions == functions assert restored.sites == (site,) + assert restored.reads == (read,) # --------------------------------------------------------------------------- keys diff --git a/tests/test_read_entry_points.py b/tests/test_read_entry_points.py new file mode 100644 index 0000000..5ad0cfc --- /dev/null +++ b/tests/test_read_entry_points.py @@ -0,0 +1,123 @@ +"""An advisory entry point reached by reading it, not only by calling it. + +Some vulnerable code runs when an attribute is read: Werkzeug parses a multipart body +when ``request.form``, ``request.files`` or ``request.data`` is read. An entry point +marked ``read`` is reached by any read of it or of an attribute of it — a subscript, an +iteration, a method call on it all read it first — while an ordinary entry point is +still reached by a call only. +""" + +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, + dump_advisories, + load_advisories, +) +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId + +PLUGINS = engine.BUNDLED_PLUGINS +PARSED = "the getter parses the whole body (commit abc1234)" + +FORM = Advisory( + "CVE-2099-0201", + "vulnlib", + "<1.1", + "reading the form parses an unbounded body", + Severity.HIGH, + entry_points=( + AdvisoryEntryPoint(SymbolId("python.vulnlib.request.form"), PARSED, read=True), + AdvisoryEntryPoint(SymbolId("python.vulnlib.request.files"), PARSED, read=True), + AdvisoryEntryPoint(SymbolId("python.vulnlib.parse"), "parse parses the body (commit abc1234)"), + ), + modules=("vulnlib",), +) + + +def analyse(root: Path, body: str, **options: object) -> engine.ProjectAnalysis: + files = { + "requirements.txt": "vulnlib==1.0\n", + "app.py": f"from vulnlib import request, parse\nimport vulnlib\n\ndef view():\n{body}", + } + for relative, text in 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((FORM,)), encoding="utf-8") + return engine.analyze_project(root, [PLUGINS], **options) # type: ignore[arg-type] + + +def reached(analysis: engine.ProjectAnalysis) -> list[tuple[int, str]]: + return [ + (f.span.start_line, f.metadata["entry_point"]) + for f in analysis.findings + if f.rule_id == "reachable-vulnerability" + ] + + +@pytest.mark.parametrize( + "body, entry", + [ + (" return request.form['name']\n", "python.vulnlib.request.form"), + (" for upload in request.files:\n pass\n", "python.vulnlib.request.files"), + (" return request.form.get('name')\n", "python.vulnlib.request.form"), + (" form = vulnlib.request.form\n return form\n", "python.vulnlib.request.form"), + ], +) +def test_reading_an_entry_point_marked_read_reaches_it(tmp_path: Path, body: str, entry: str) -> None: + assert reached(analyse(tmp_path, body)) == [(5, entry)] + + +def test_several_reads_of_one_entry_point_in_a_function_are_one_finding(tmp_path: Path) -> None: + body = " first = request.form['a']\n second = request.form['b']\n return first, request.form.get('c')\n" + + assert reached(analyse(tmp_path, body)) == [(5, "python.vulnlib.request.form")] + + +def test_calling_an_entry_point_marked_read_is_one_finding(tmp_path: Path) -> None: + assert reached(analyse(tmp_path, " return request.files()\n")) == [(5, "python.vulnlib.request.files")] + + +def test_an_entry_point_not_marked_read_is_still_reached_by_a_call_only(tmp_path: Path) -> None: + assert reached(analyse(tmp_path / "referenced", " handler = parse\n return handler\n")) == [] + assert reached(analyse(tmp_path / "called", " return parse(request)\n")) == [(5, "python.vulnlib.parse")] + + +def test_a_module_served_from_the_cache_still_reaches_what_it_reads(tmp_path: Path) -> None: + cache = ProjectCache(tmp_path / "cache") + body = " return request.form['name']\n" + + first = analyse(tmp_path / "src", body, cache=cache) + second = analyse(tmp_path / "src", body, cache=cache) + + assert "app" in second.reused + assert reached(second) == reached(first) == [(5, "python.vulnlib.request.form")] + + +def test_read_entry_points_round_trip_through_advisory_files(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((FORM,)), encoding="utf-8") + + assert load_advisories(path) == (FORM,) + entries = json.loads(path.read_text(encoding="utf-8"))["advisories"][0]["entry_points"] + assert [entry.get("read") for entry in entries] == [True, True, None] + + +def test_a_malformed_read_flag_is_rejected(tmp_path: Path) -> None: + document = json.loads(dump_advisories((FORM,))) + document["advisories"][0]["entry_points"][0]["read"] = "yes" + path = tmp_path / "advisories.json" + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError, match="read"): + load_advisories(path) From 6895ea216ef3a325cafd1ccace1cec6ba4c53b25 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 24 Sep 2026 17:53:36 +0200 Subject: [PATCH 2/2] feat: reach advisory entry points marked read through attribute reads --- CHANGELOG.md | 8 +++ docs/plugins.md | 7 +- docs/usage.md | 12 +++- .../reachable_vulnerability.py | 71 ++++++++++++++----- src/coretrace_python/cache.py | 18 +++-- src/coretrace_python/dependency/advisories.py | 24 ++++--- src/coretrace_python/dependency/graph.py | 7 +- src/coretrace_python/engine.py | 15 +++- .../interprocedural/__init__.py | 2 + .../interprocedural/callgraph.py | 40 ++++++++++- 10 files changed, 164 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f9f334..f020fa4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ - `yaml.load_all`, `yaml.full_load_all` and `yaml.unsafe_load_all` are deserialization sinks like their single-document versions; `yaml.load_all` with a safe loader is not, and `yaml.safe_load_all` stays safe (#123). +### Dependencies + +- An advisory entry point marked `read` is an attribute whose getter runs the affected code, such as `request.form`: reading it is reachable, called or not — a subscript, an iteration or a method called on it reads it — with one finding per function, where it is first read (#111). + +### Plugins + +- The call graph records the symbols each function reads, called or not, where it first reads each: `CallGraph.reads(function)` gives them as `SymbolRead` records, modules served from the cache keep them, and the cache format is bumped (#111). + ### Precision - A sink call can be made safe by one of its arguments: a `SafeArgument` model names the argument and the values that take kinds off the sink, only when the call gives one explicitly. `yaml.load` with `SafeLoader`, `BaseLoader` or their C versions, under every spelling, is no longer an insecure deserialization; an absent loader, another loader, a variable or unpacked arguments still are (#121). diff --git a/docs/plugins.md b/docs/plugins.md index aa11065..e766074 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -227,8 +227,11 @@ contributed, the policy, the project root, and each module's imports, call graph functions. `ctx.functions(module)` gives every function as the call graph names it (`Admin.get`, `outer.inner`, ``), with its span, so a finding can be placed in one, and the label of the entry point it is (`http` for a route, `argv` for a command) -when the models make it one. The shipped `vulnerable-dependency`, -`reachable-vulnerability` and `dependency-policy` plugins are of this kind. +when the models make it one. `ctx.call_graph(module)` gives each function's call sites +(`sites(function)`) and the symbols it reads, called or not (`reads(function)`: reading +`request.form['name']` reads `python.flask.request.form`), each where it first reads it. +The shipped `vulnerable-dependency`, `reachable-vulnerability` and `dependency-policy` +plugins are of this kind. ```python from collections.abc import Sequence diff --git a/docs/usage.md b/docs/usage.md index 67dda5d..3894420 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -151,7 +151,7 @@ sources and in configuration files. | Rule | Trigger | |---|---| | `vulnerable-dependency` | A requirement allows a version affected by an advisory, reported at its line. | -| `reachable-vulnerability` | A call in the project to an API the advisory affects. | +| `reachable-vulnerability` | A call in the project to an API the advisory affects, or a read of an entry point marked `read`. | | `exploitable-vulnerability` | Attacker-controlled data reaches such a call. Critical. | | `denied-dependency` | A package the policy denies. | | `unpinned-dependency` | A requirement without an exact pin when the policy requires pins. | @@ -291,6 +291,13 @@ call to either kind of symbol is reachable. `modules` names the top-level module package installs, so the analyzer can tell an imported package from a merely required one; it defaults to the package name. +An entry point marked `"read": true` is an attribute whose getter runs the affected code, +such as a request body parsed on first access +(`{"symbol": "python.flask.request.form", "justification": "…", "read": true}`): reading +it is reachable, called or not — `request.form['name']`, `for key in request.form`, +`request.form.get('name')` all read it. A function reading it several times is reported +once, where it first reads it. + 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 @@ -305,7 +312,8 @@ template's origin — cannot be decided and is always pending review, never drop 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). The last two carry +symbol is called, or an entry point marked `read` is read) 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 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 96b83b3..9f11f88 100644 --- a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py +++ b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -1,16 +1,24 @@ -"""Calls, anywhere in the project, to an API affected by a vulnerable requirement.""" +"""Calls, anywhere in the project, to an API affected by a vulnerable requirement, and +reads of an affected attribute whose getter runs the vulnerable code.""" from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Iterator, Sequence from typing import ClassVar from coretrace_python.analysis import AnyAnalysis -from coretrace_python.dependency import DependencyAnalysis -from coretrace_python.dependency.correlation import affected_symbols, check_conditions, evidence +from coretrace_python.dependency import Advisory, DependencyAnalysis +from coretrace_python.dependency.correlation import ( + ConditionCheck, + 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 +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceSpan class ReachableVulnerabilityPlugin(ProjectPlugin): @@ -29,21 +37,46 @@ 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) + entry = advisory.entry_point(site.target.symbol) + if entry is not None and entry.read: + continue # a call reads its callee first: reported with the reads + check = check_conditions(entry, site.arguments) if check.contradicted is not None: continue - findings.append( - Finding( - rule_id="reachable-vulnerability", - message=( - f"{advisory.id}: {site.target.symbol} is affected in the required " - f"{advisory.package} {advisory.vulnerable}: {advisory.summary}" - ), - severity=advisory.severity, - confidence=Confidence.HIGH, - span=site.location, - function=function, - metadata=evidence(advisory, site.target.symbol, "reachable", check), - ) - ) + findings.append(_reached(advisory, site.target.symbol, site.location, function, check)) + # Reading ``request.form.get`` reads ``request.form``: one finding per entry + # point and function, where the function first reads it. + reported: set[tuple[Advisory, SymbolId]] = set() + for read in graph.reads(function): + for symbol in _read_through(read.symbol): + for advisory in affected.get(symbol, ()): + entry = advisory.entry_point(symbol) + if entry is None or not entry.read or (advisory, symbol) in reported: + continue + reported.add((advisory, symbol)) + check = check_conditions(entry, None) + findings.append(_reached(advisory, symbol, read.location, function, check)) return findings + + +def _read_through(symbol: SymbolId) -> Iterator[SymbolId]: + """``symbol`` and every symbol it is an attribute of, as reading it reads them all.""" + + components = symbol.canonical_name.split(".") + for end in range(len(components), 1, -1): + yield SymbolId(".".join(components[:end])) + + +def _reached(advisory: Advisory, symbol: SymbolId, span: SourceSpan, function: str, check: ConditionCheck) -> Finding: + return Finding( + rule_id="reachable-vulnerability", + message=( + f"{advisory.id}: {symbol} is affected in the required " + f"{advisory.package} {advisory.vulnerable}: {advisory.summary}" + ), + severity=advisory.severity, + confidence=Confidence.HIGH, + span=span, + function=function, + metadata=evidence(advisory, symbol, "reachable", check), + ) diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index dd3aa91..1baf0ec 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -4,9 +4,10 @@ source text and identity, the engine, schema and plugin API versions, the plugins and their code, the security models, the advisories, the dependency graph, and the keys of the project modules it imports transitively. A module whose key is unchanged on a later -run is served from the cache: its summaries seed the project index, its call sites serve -the project plugins and its findings are reported as they were. Entries are JSON, so a -tampered or foreign file can never execute anything; an unreadable entry is a miss. +run is served from the cache: its summaries seed the project index, its call sites and +symbol reads serve the project plugins and its findings are reported as they were. +Entries are JSON, so a tampered or foreign file can never execute anything; an +unreadable entry is a miss. """ from __future__ import annotations @@ -33,13 +34,14 @@ Mutation, NonlocalWrite, SummaryIndex, + SymbolRead, Target, UnknownTarget, ) from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan -CACHE_FORMAT = 7 +CACHE_FORMAT = 8 @dataclass(frozen=True) @@ -50,6 +52,7 @@ class CachedModule: summaries: Mapping[str, FunctionSummary] sites: tuple[CallSite, ...] findings: tuple[Finding, ...] + reads: tuple[SymbolRead, ...] = () def __post_init__(self) -> None: object.__setattr__(self, "summaries", MappingProxyType(dict(self.summaries))) @@ -134,6 +137,7 @@ def encode(module: CachedModule) -> dict[str, Any]: "summaries": {name: _encode_summary(s) for name, s in module.summaries.items()}, "sites": [_encode_site(site) for site in module.sites], "findings": [_encode_finding(finding) for finding in module.findings], + "reads": [[r.function, _encode_span(r.location), str(r.symbol)] for r in module.reads], } @@ -145,6 +149,7 @@ def decode(data: Mapping[str, Any]) -> CachedModule: {_string(name): _decode_summary(s) for name, s in data["summaries"].items()}, tuple(_decode_site(site) for site in data["sites"]), tuple(_decode_finding(finding) for finding in data["findings"]), + tuple(_decode_read(read) for read in data["reads"]), ) @@ -194,6 +199,11 @@ def _decode_function(data: Any) -> ModuleFunction: return ModuleFunction(_string(name), _decode_span(span), None if entry_point is None else _string(entry_point)) +def _decode_read(data: Any) -> SymbolRead: + function, span, symbol = data + return SymbolRead(_string(function), _decode_span(span), SymbolId(_string(symbol))) + + def _encode_finding(finding: Finding) -> dict[str, Any]: return { "rule": finding.rule_id, diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index 99b8c9f..edfa4fd 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -138,14 +138,7 @@ def dump_advisories(advisories: Iterable[Advisory]) -> str: "severity": a.severity.value, "affected_symbols": [str(s) for s in a.affected_symbols], "aliases": list(a.aliases), - "entry_points": [ - { - "symbol": str(e.symbol), - "justification": e.justification, - "conditions": [_condition_entry(c) for c in e.conditions], - } - for e in a.entry_points - ], + "entry_points": [_entry_point_entry(e) for e in a.entry_points], "modules": list(a.modules), } for a in advisories @@ -154,6 +147,17 @@ def dump_advisories(advisories: Iterable[Advisory]) -> str: return json.dumps(document, indent=2) + "\n" +def _entry_point_entry(entry_point: AdvisoryEntryPoint) -> dict[str, Any]: + entry: dict[str, Any] = { + "symbol": str(entry_point.symbol), + "justification": entry_point.justification, + "conditions": [_condition_entry(c) for c in entry_point.conditions], + } + if entry_point.read: + entry["read"] = True + return entry + + def _condition_entry(condition: Condition) -> dict[str, Any]: entry: dict[str, Any] = {"kind": condition.kind, "text": condition.text} if condition.argument is not None: @@ -194,10 +198,14 @@ def _advisory(entry: Mapping[str, Any]) -> Advisory: def _entry_point(entry: Mapping[str, Any]) -> AdvisoryEntryPoint: + read = entry.get("read", False) + if not isinstance(read, bool): + raise TypeError(f"entry point read must be true or false, got {read!r}") return AdvisoryEntryPoint( SymbolId(str(entry["symbol"])), str(entry["justification"]), tuple(_condition(c) for c in entry.get("conditions") or []), + read, ) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 3450ec2..cc82193 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -168,11 +168,16 @@ def checkable(self) -> bool: @dataclass(frozen=True) class AdvisoryEntryPoint: """A public API through which a project reaches an affected symbol, justified by the - fixing commit or by a call path, with the conditions under which it is affected.""" + fixing commit or by a call path, with the conditions under which it is affected. + + An entry point is reached by a call, or — when ``read`` is set, for an attribute whose + getter runs the affected code, such as a lazily parsed request body — by any read of + it or of an attribute of it.""" symbol: SymbolId justification: str conditions: tuple[Condition, ...] = () + read: bool = False DIRECT = "affected symbol, changed by the fix" diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index b574597..0be43f4 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -76,6 +76,7 @@ ProjectSummaries, SummaryAnalysis, SummaryIndex, + SymbolRead, build_module_graph, discover_sources, project_symbol, @@ -466,7 +467,15 @@ def analyze_project( sites: dict[str, list[CallSite]] = {function.name: [] for function in entry.functions} 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()) + reads: dict[str, list[SymbolRead]] = {} + for read in entry.reads: + reads.setdefault(read.function, []).append(read) + call_graphs[name] = CallGraph( + {}, + {f: tuple(s) for f, s in sites.items()}, + frozenset(), + reads={f: tuple(r) for f, r in reads.items()}, + ) context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs, policy, root, functions) for plugin in all_plugins: if isinstance(plugin, ProjectPlugin): @@ -651,7 +660,8 @@ def _configuration_key( def _analyse_module( manager: AnalysisManager, plugins: tuple[Plugin, ...], affected: Affected ) -> CachedModule: - """One module's findings, summaries and call sites: what the cache keeps (§11).""" + """One module's findings, summaries, call sites and symbol reads: what the cache + keeps (§11).""" findings, supported = _check_module(manager, plugins) graph = manager.get(CallGraphAnalysis) @@ -671,6 +681,7 @@ def _analyse_module( _summaries_of(manager), tuple(site for function in graph.functions for site in graph.sites(function)), (*findings, *correlated), + tuple(read for function in graph.functions for read in graph.reads(function)), ) diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index e2ed9cd..d6bd03b 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -8,6 +8,7 @@ ExternalSymbol, KnownFunction, ModuleFunction, + SymbolRead, Target, UnknownTarget, ) @@ -46,6 +47,7 @@ "SummaryAnalysis", "SummaryIndex", "SummaryTable", + "SymbolRead", "Target", "UnknownTarget", "build_module_graph", diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index a076d10..c93e810 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -3,7 +3,8 @@ Every call site of every analysable function resolves, through the SSA form, to a ``KnownFunction`` defined at module level, an ``ExternalSymbol`` reached through imports or builtins, or ``UnknownTarget`` (parameters, attributes, methods) until type inference -and framework models narrow it down. +and framework models narrow it down. The graph also keeps the symbols each function +reads, called or not: reading ``request.form`` runs its getter. """ from __future__ import annotations @@ -93,6 +94,16 @@ class CallSite: arguments: Arguments = field(default_factory=Arguments) +@dataclass(frozen=True) +class SymbolRead: + """Where ``function`` first reads ``symbol``: an imported name or an attribute of a + value denoting a symbol (``request.form`` in ``request.form['name']``).""" + + function: str + location: SourceSpan + symbol: SymbolId + + @dataclass(frozen=True) class ModuleFunction: """One function of a module as a project plugin sees it: the name the call graph @@ -111,8 +122,10 @@ def __init__( sites: Mapping[str, tuple[CallSite, ...]], unsupported: frozenset[str], symbols: Mapping[str, Mapping[Value, SymbolId]] | None = None, + reads: Mapping[str, tuple[SymbolRead, ...]] | None = None, ) -> None: self._symbols = {name: MappingProxyType(dict(found)) for name, found in (symbols or {}).items()} + self._reads = MappingProxyType(dict(reads or {})) self.definitions: Mapping[str, nodes.Function] = MappingProxyType(dict(definitions)) # A graph rebuilt from cached call sites has sites but no definitions. self.functions = tuple(dict.fromkeys((*definitions, *sites))) @@ -138,6 +151,11 @@ def symbols(self, name: str) -> Mapping[Value, SymbolId]: def sites(self, caller: str) -> tuple[CallSite, ...]: return self._sites.get(caller, ()) + def reads(self, function: str) -> tuple[SymbolRead, ...]: + """The symbols ``function`` reads, each where it first reads it.""" + + return self._reads.get(function, ()) + def target_at(self, caller: str, location: SourceSpan) -> Target: site = self._at.get((caller, location)) return site.target if site is not None else UnknownTarget() @@ -219,6 +237,22 @@ def denoted(value: Value) -> str | None: ) +def _reads(name: str, function: FunctionIR, symbols: Mapping[Value, SymbolId]) -> tuple[SymbolRead, ...]: + first: dict[SymbolId, SymbolRead] = {} + for block in function.blocks: + for instruction in block.instructions: + if isinstance(instruction, Symbol | GetAttr) and instruction.result in symbols: + read = SymbolRead(name, instruction.location, symbols[instruction.result]) + known = first.get(read.symbol) + if known is None or _position(read.location) < _position(known.location): + first[read.symbol] = read + return tuple(sorted(first.values(), key=lambda read: (_position(read.location), str(read.symbol)))) + + +def _position(span: SourceSpan) -> tuple[int, int]: + return span.start_line, span.start_column + + def resolve_targets( function: FunctionIR, scopes: ScopeTable, @@ -371,6 +405,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: break sites: dict[str, tuple[CallSite, ...]] = {} symbols: dict[str, Mapping[Value, SymbolId]] = {} + reads: dict[str, tuple[SymbolRead, ...]] = {} unsupported: set[str] = set() for name, function in definitions.items(): try: @@ -407,4 +442,5 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: ) ) sites[name] = tuple(found) - return CallGraph(definitions, sites, frozenset(unsupported), symbols) + reads[name] = _reads(name, ssa, symbols[name]) + return CallGraph(definitions, sites, frozenset(unsupported), symbols, reads)