mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
refactor(p1): one source-facts module owns function ranges, callers, declaration tables and language aliases
src/openultrasast/source.py replaces the copies in semantic/functions.py, plane/tasks/repo_facts.py, learn/features.py, learn/excerpt.py and plane/tasks/verify.py. Two engines share one sentinel (<module>), one DECLARATION/KEYWORDS/RESERVED set and one alias map: the parser engine (ranges, enclosing, bounds, call_sites, callers_of, facts_for) for the scorer, the overlay and the hunter's annotations; the line-pattern engine (declared_functions, enclosing_at, function_span, function_bounds, the brace matcher) that labels, features, roles, excerpts and plane facts were built on. learn no longer imports plane.tasks. Frozen records that store the old sentinel are mapped on read (module_function). Gate: labels rebuilt before and after are identical, 7,921 rows, after sentinel normalisation (benchmarks/measurements/2026-10-13-source-module- label-diff/record.json). 490 targeted tests pass; 3,312 collect. Net: 657 lines deleted, 212 added. benchmarks/unseen keeps <global> for its own frozen manifests. Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
f63703a604
commit
062a7b780e
+3
-1
@@ -198,7 +198,9 @@ lines = 4803
|
||||
|
||||
[capability.core]
|
||||
status = "active"
|
||||
modules = ["cli.py", "config.py", "contracts.py", "preprocess.py", "findings.py", "rank.py", "mapping.py", "ruleset", "policy", "reports.py", "stages.py", "run.py", "provenance.py", "profiling.py"]
|
||||
modules = ["cli.py", "config.py", "contracts.py", "preprocess.py", "source.py", "findings.py", "rank.py", "mapping.py", "ruleset", "policy", "reports.py", "stages.py", "run.py", "provenance.py", "profiling.py"]
|
||||
# source.py (P1, 2026-10-13) owns function ranges, enclosing function, bounds, call sites, product files, declaration
|
||||
# tables and language aliases; learn no longer imports plane.tasks
|
||||
note = "hunter.py is the regex quick-rule pool and should be renamed"
|
||||
|
||||
[capability.pairs-and-benchmarks]
|
||||
|
||||
@@ -85,7 +85,7 @@ def php_code(text: str) -> str:
|
||||
"""`text` with comments, string contents and inline HTML blanked to spaces, newlines kept (line numbers hold).
|
||||
|
||||
Enough to find function bodies by brace matching; heredoc/nowdoc bodies are blanked too. Not a parser: a file
|
||||
that defeats it yields no function, and the site falls back to `<global>`, which the freeze record shows."""
|
||||
that defeats it yields no function, and the site falls back to `<module>`, which the freeze record shows."""
|
||||
out, i, n, php = list(text), 0, len(text), False
|
||||
|
||||
def blank(start: int, end: int) -> None:
|
||||
@@ -148,10 +148,10 @@ def php_functions(text: str) -> list[tuple[str, int, int]]:
|
||||
|
||||
|
||||
def enclosing(spans: list[tuple[str, int, int]], line: int, inserted: bool = False) -> str:
|
||||
"""The innermost named function containing `line`, or `<global>`. For a pure insertion, `line` is the line the
|
||||
"""The innermost named function containing `line`, or `<module>`. For a pure insertion, `line` is the line the
|
||||
new code follows, so a function whose closing brace is on that line does not contain it."""
|
||||
inside = [(last - first, name) for name, first, last in spans if first <= line <= last and not (inserted and line == last)]
|
||||
return min(inside)[1] if inside else "<global>"
|
||||
return min(inside)[1] if inside else "<module>"
|
||||
|
||||
|
||||
def old_hunks(repo: Path, vulnerable: str, fixed: str, path: str) -> list[tuple[int, int, bool]]:
|
||||
@@ -166,14 +166,14 @@ def old_hunks(repo: Path, vulnerable: str, fixed: str, path: str) -> list[tuple[
|
||||
|
||||
|
||||
def changed_functions(repo: Path, vulnerable: str, fixed: str, path: str) -> list[str]:
|
||||
"""The functions (or `<global>`) of `path` at the vulnerable pin that the fix's hunks touch, in hunk order."""
|
||||
"""The functions (or `<module>`) of `path` at the vulnerable pin that the fix's hunks touch, in hunk order."""
|
||||
text = (blob(repo, vulnerable, path) or b"").decode(errors="replace")
|
||||
spans, names = php_functions(text), []
|
||||
for first, last, inserted in old_hunks(repo, vulnerable, fixed, path):
|
||||
for line in range(first, last + 1):
|
||||
name = enclosing(spans, line, inserted)
|
||||
# New code added between functions (a new helper) changes no vulnerable-side code: no site.
|
||||
if name not in names and not (inserted and name == "<global>"):
|
||||
if name not in names and not (inserted and name == "<module>"):
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
@@ -198,12 +198,12 @@ def resolve_sites(repo: Path, case: dict[str, object], changed: list[str]) -> di
|
||||
|
||||
|
||||
def fix_changes_sites(repo: Path, case: dict[str, object], sites: list[str]) -> dict[str, bool]:
|
||||
"""Per declared site: does the fix's diff change a line inside that function (or the file, for `<global>`)?"""
|
||||
"""Per declared site: does the fix's diff change a line inside that function (or the file, for `<module>`)?"""
|
||||
result = {}
|
||||
for site in sites:
|
||||
path, _, name = site.partition("::")
|
||||
hunks = old_hunks(repo, str(case["vulnerable"]), str(case["fixed"]), path)
|
||||
if name == "<global>" or not hunks:
|
||||
if name == "<module>" or not hunks:
|
||||
result[site] = bool(hunks)
|
||||
continue
|
||||
spans = php_functions((blob(repo, str(case["vulnerable"]), path) or b"").decode(errors="replace"))
|
||||
@@ -289,7 +289,7 @@ def verify(case: dict[str, object], cache: Path) -> dict[str, object]:
|
||||
# Declared `path::function` sites (population v2) are what the verification reads; the first names the
|
||||
# sink file, and every function declared in that file must appear in it.
|
||||
sink_file = sites[0].partition("::")[0]
|
||||
functions = [site.partition("::")[2] for site in sites if site.partition("::")[0] == sink_file and not site.endswith("::<global>")]
|
||||
functions = [site.partition("::")[2] for site in sites if site.partition("::")[0] == sink_file and not site.endswith("::<module>")]
|
||||
else:
|
||||
sink_file, functions = sink_parts(str(case["sink"]))
|
||||
record["sink_file"], record["sink_functions"] = sink_file, functions
|
||||
|
||||
@@ -76,10 +76,10 @@ from openultrasast.plane.harvest import declares
|
||||
from openultrasast.plane.tasks.alerts import covers
|
||||
from openultrasast.plane.tasks.alerts import engine_languages as _engine_languages
|
||||
from openultrasast.plane.tasks.alerts import quick_languages as _quick_languages
|
||||
from openultrasast.plane.tasks.repo_facts import GLOBAL
|
||||
from openultrasast.preprocess import detect_language
|
||||
from openultrasast.rank import rank_targets
|
||||
from openultrasast.ruleset import DEFAULT_RULESET_DIR, load_ruleset
|
||||
from openultrasast.source import MODULE_FUNCTION
|
||||
|
||||
RESULTS = Path.home() / "ousast-results"
|
||||
MISSING = Part("failed", version="missing:not-run")
|
||||
@@ -306,7 +306,7 @@ def static_parts(root: Path, index: SourceIndex, quick: Mapping, engine: Mapping
|
||||
parts["roles"] = roles_part(function_of(lines, path, language, function), inferred, language)
|
||||
parts["delta"] = NOT_APPLICABLE
|
||||
if repository and entries is not None:
|
||||
distance = 0 if function == GLOBAL and path in entries[1] else entry_distance(function, entries[0], index.callers())
|
||||
distance = 0 if function == MODULE_FUNCTION and path in entries[1] else entry_distance(function, entries[0], index.callers())
|
||||
parts["entry_points"] = Part("ran", {"facts.entry_distance": distance})
|
||||
parts["facts"] = facts_part(_callers_of(index, index.callers(), path, function), function, "host")
|
||||
else:
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
"""Compare the baseline labels (old modules) with the labels built on source.py, sentinel-normalised."""
|
||||
import glob, json, collections, sys
|
||||
def load(pattern):
|
||||
files = glob.glob(pattern)
|
||||
if not files:
|
||||
sys.exit(f"no labels file for {pattern}")
|
||||
rows = [json.loads(l) for l in open(files[0]) if l.strip()]
|
||||
return files[0], rows
|
||||
bfile, before = load("/home/mc/ousast-results/fossil/p1-baseline/out/labels-*.jsonl")
|
||||
afile, after = load("/home/mc/ousast-results/fossil/p1-after/out/labels-*.jsonl")
|
||||
def key(r):
|
||||
s = json.dumps(r, sort_keys=True).replace("<global>", "<module>")
|
||||
return s
|
||||
B = collections.Counter(key(r) for r in before); A = collections.Counter(key(r) for r in after)
|
||||
only_b = B - A; only_a = A - B
|
||||
print("baseline:", bfile, len(before), "rows"); print("after: ", afile, len(after), "rows")
|
||||
print("identical after sentinel normalisation:", len(only_b) == 0 and len(only_a) == 0)
|
||||
print("rows only in baseline:", sum(only_b.values()), " rows only in after:", sum(only_a.values()))
|
||||
def summarize(c, title):
|
||||
print(f"--- {title} (first 8)")
|
||||
for s, n in list(c.items())[:8]:
|
||||
r = json.loads(s); print(" ", {k: r.get(k) for k in ("candidate", "family", "label", "source", "unit", "pin_role") if k in r})
|
||||
summarize(only_b, "only baseline"); summarize(only_a, "only after")
|
||||
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"date": "2026-10-13",
|
||||
"what": "Label diff gate for P1 (PATHFINDER-2026-10-13/04-handoff-prompts.md): labels built by `ousast learn labels` with the modules before source.py (HEAD f63703a, exported pristine) and after, compared row by row after mapping the legacy module-level sentinel <global> to <module>.",
|
||||
"before": {
|
||||
"file": "labels-c3d9c302eadf79de.jsonl",
|
||||
"rows": 7921,
|
||||
"sha256": "c3d9c302eadf79dedd3b91b5545ee4f2d91d634dafe244eb7f7cf73e76d3f1f4",
|
||||
"code": "f63703a"
|
||||
},
|
||||
"after": {
|
||||
"file": "labels-7701cc7339a0d0f7.jsonl",
|
||||
"rows": 7921,
|
||||
"sha256": "7701cc7339a0d0f7aff036581c8417f5afc645d9e45bc067787fbf767425f02a",
|
||||
"code": "working tree before the P1 commit"
|
||||
},
|
||||
"result": "identical after sentinel normalisation (0 rows only-before, 0 rows only-after)",
|
||||
"iterations": [
|
||||
"1: Codex's parser-first rewrite of the learn and plane modules: 18 test failures (function bounds semantics), reverted those modules to the line-pattern engine moved into source.py",
|
||||
"2: 4 rows missing: KEYWORDS had been widened to the parser's reserved set; restored the 22-word set (no change in output, kept for correctness)",
|
||||
"3: 4 rows still missing: frozen population records store declared sites as path::<global>; normalised on read (module_function) in labels and in function_bounds/function_span",
|
||||
"4: identical"
|
||||
],
|
||||
"sources": "src/openultrasast/learn/sources.toml (population-v1, population-v2, dev-php, pairs, advisory-fixes, ...; v3 excluded unread)",
|
||||
"note": "The label files themselves stay outside the repo (they name functions in private repositories): ~/ousast-results/fossil/p1-{baseline,after}/out/."
|
||||
}
|
||||
@@ -106,10 +106,10 @@ def _body(lines: list[str], span: tuple[int, int]) -> str:
|
||||
|
||||
|
||||
def extract(source: str, language: str, function: str) -> tuple[list[str], tuple[int, int]] | None:
|
||||
from openultrasast.learn.excerpt import LANGUAGE_ALIASES, function_bounds
|
||||
from openultrasast.source import canonical_language, function_bounds
|
||||
|
||||
lines = source.splitlines()
|
||||
span = function_bounds(lines, LANGUAGE_ALIASES.get(language, language), function)
|
||||
span = function_bounds(lines, canonical_language(language), function)
|
||||
return (lines, span) if span else None
|
||||
|
||||
|
||||
@@ -159,14 +159,14 @@ def materialise(pair: dict[str, Any], *, force: bool = False) -> dict[str, Any]:
|
||||
|
||||
def verify(pair: dict[str, Any]) -> dict[str, Any]:
|
||||
"""The label builder's check, from the cached excerpts: declared on both sides, body changed, a real fix."""
|
||||
from openultrasast.learn.excerpt import LANGUAGE_ALIASES, function_bounds
|
||||
from openultrasast.source import canonical_language, function_bounds
|
||||
|
||||
result: dict[str, Any] = {"name": pair["name"], "ok": False, "bytes_read": 0}
|
||||
commit, parent = str(pair.get("commit", "")), str(pair.get("parent", ""))
|
||||
if not (_SHA.fullmatch(commit) and _SHA.fullmatch(parent)) or commit == parent:
|
||||
result["reason"] = "fix commit must be a full SHA distinct from its parent"
|
||||
return result
|
||||
language = LANGUAGE_ALIASES.get(str(pair["language"]), str(pair["language"]))
|
||||
language = canonical_language(str(pair["language"]))
|
||||
spans = {}
|
||||
for side, path in zip(("vuln", "fixed"), excerpt_paths(pair), strict=True):
|
||||
if not path.is_file():
|
||||
|
||||
@@ -15,7 +15,8 @@ import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from openultrasast.semantic.functions import ANON_FUNCTION, named_function_ranges
|
||||
from openultrasast.source import ANON_FUNCTION
|
||||
from openultrasast.source import ranges as source_ranges
|
||||
except ImportError as exc: # never degrade into "no named function": that path deletes corpora under --prune
|
||||
raise SystemExit(f"catalog_gen needs the openultrasast package (run it with the project venv): {exc}") from exc
|
||||
|
||||
@@ -268,7 +269,7 @@ def _language(recipe: dict[str, object]) -> str:
|
||||
|
||||
|
||||
def _ranges(path: Path, language: str) -> tuple[tuple[str, int, int], ...] | None:
|
||||
ranges = named_function_ranges(str(path), path.read_text(errors="ignore"), language)
|
||||
ranges = source_ranges(str(path), path.read_text(errors="ignore"), language)
|
||||
if ranges is None:
|
||||
return None # parser rejected the excerpt; a regex label would not scope
|
||||
return tuple(r for r in ranges if r[0] != ANON_FUNCTION)
|
||||
|
||||
@@ -6,6 +6,7 @@ from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
from .preprocess import enumerate_source_files
|
||||
from .source import callers_of, product_files
|
||||
|
||||
|
||||
class PathEscapesRepo(ValueError):
|
||||
@@ -81,15 +82,13 @@ def grep_repo(root: Path, pattern: str, *, max_matches: int) -> list[dict[str, o
|
||||
|
||||
|
||||
def find_refs(root: Path, symbol: str) -> list[dict[str, object]]:
|
||||
resolved_root = root.resolve()
|
||||
refs: list[dict[str, object]] = []
|
||||
if symbol:
|
||||
compiled = re.compile(rf"\b{re.escape(symbol)}\b")
|
||||
for relative, lines in _source_lines(resolved_root):
|
||||
for line_no, line in enumerate(lines, start=1):
|
||||
if compiled.search(line):
|
||||
refs.append({"path": relative, "name": symbol, "line": line_no, "text": line[:_LINE_CHARS], "source": "text"})
|
||||
return refs
|
||||
if not symbol:
|
||||
return []
|
||||
lines = dict(_source_lines(root))
|
||||
return [
|
||||
{"path": path, "name": symbol, "line": line, "text": lines[path][line - 1][:_LINE_CHARS], "source": "text"}
|
||||
for path, line in callers_of(root, symbol, files=product_files(root))
|
||||
]
|
||||
|
||||
|
||||
def _iter_clamped_source_files(root: Path) -> Iterator[tuple[str, Path]]:
|
||||
|
||||
@@ -32,16 +32,17 @@ import re
|
||||
from collections.abc import Iterable, Sequence
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ..plane.tasks.repo_facts import DECLARATION, GLOBAL, _declared_name
|
||||
from .features import function_span
|
||||
from ..source import ( # noqa: F401 (re-exported: labels, examples and harvest import them from here)
|
||||
HASH_COMMENTS,
|
||||
SLASH_COMMENTS,
|
||||
_comment_ranges,
|
||||
canonical_language,
|
||||
function_bounds,
|
||||
resolve_language,
|
||||
)
|
||||
from .labels import _ADVISORY, security_reason
|
||||
|
||||
REDACTED = "<redacted>"
|
||||
LANGUAGE_ALIASES = {"c_cpp": "cpp", "c++": "cpp", "js": "javascript", "ts": "typescript", "py": "python"}
|
||||
HASH_COMMENTS = frozenset({"python", "php", "ruby", "shell", "perl", "yaml", "r"})
|
||||
SLASH_COMMENTS = frozenset(
|
||||
{"c", "cpp", "java", "javascript", "typescript", "php", "go", "groovy", "csharp", "kotlin", "rust", "swift", "scala"}
|
||||
) # fmt: skip
|
||||
MAX_LINE_CHARS = 240
|
||||
_HEX = re.compile(r"\b[0-9a-f]{7,40}\b", re.IGNORECASE)
|
||||
|
||||
@@ -87,55 +88,6 @@ def excerpt_sha(text: str) -> str:
|
||||
# --- comments ---------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _comment_ranges(line: str, state: str | None, language: str) -> tuple[list[tuple[int, int]], str | None]:
|
||||
"""The comment spans of one line and the block state carried to the next (``block`` inside ``/* */``, a triple
|
||||
quote inside a Python docstring). Strings are skipped so a ``#`` in a literal is not a comment."""
|
||||
hashes, slashes, python = language in HASH_COMMENTS, language in SLASH_COMMENTS, language == "python"
|
||||
ranges: list[tuple[int, int]] = []
|
||||
i, n, quote = 0, len(line), None
|
||||
if state is not None:
|
||||
close = "*/" if state == "block" else state
|
||||
j = line.find(close)
|
||||
if j < 0:
|
||||
return [(0, n)], state
|
||||
ranges.append((0, j + len(close)))
|
||||
i, state = j + len(close), None
|
||||
while i < n:
|
||||
c = line[i]
|
||||
if quote:
|
||||
if c == "\\":
|
||||
i += 2
|
||||
continue
|
||||
if c == quote:
|
||||
quote = None
|
||||
i += 1
|
||||
continue
|
||||
if python and line.startswith(('"""', "'''"), i) and not line[:i].strip():
|
||||
delim = line[i : i + 3]
|
||||
j = line.find(delim, i + 3)
|
||||
if j < 0:
|
||||
ranges.append((i, n))
|
||||
return ranges, delim
|
||||
ranges.append((i, j + 3))
|
||||
i = j + 3
|
||||
continue
|
||||
if c in "\"'`":
|
||||
quote = c
|
||||
elif (slashes and line.startswith("//", i)) or (hashes and c == "#"):
|
||||
ranges.append((i, n))
|
||||
break
|
||||
elif slashes and line.startswith("/*", i):
|
||||
j = line.find("*/", i + 2)
|
||||
if j < 0:
|
||||
ranges.append((i, n))
|
||||
return ranges, "block"
|
||||
ranges.append((i, j + 2))
|
||||
i = j + 2
|
||||
continue
|
||||
i += 1
|
||||
return ranges, state
|
||||
|
||||
|
||||
_OPEN = ("/**", "/*", "//", "#", '"""', "'''", "*")
|
||||
_CLOSE = ("*/", '"""', "'''")
|
||||
|
||||
@@ -153,7 +105,7 @@ def _redact_comment(text: str) -> str:
|
||||
def redact(lines: Sequence[str], language: str, identities: Iterable[str] = ()) -> list[str]:
|
||||
"""``lines`` with security wording in comments and advisory ids anywhere replaced by ``<redacted>``, and every
|
||||
identity string (case-insensitive, at least 4 characters) replaced too."""
|
||||
language = LANGUAGE_ALIASES.get(language, language)
|
||||
language = canonical_language(language)
|
||||
names = sorted({i for i in identities if len(i) >= 4}, key=len, reverse=True)
|
||||
identity = re.compile("|".join(re.escape(n) for n in names), re.IGNORECASE) if names else None
|
||||
out: list[str] = []
|
||||
@@ -188,135 +140,6 @@ def identities_of(repo: str = "", path: str = "", commits: Iterable[str] = ()) -
|
||||
# --- spans and excerpts -----------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
_UNKNOWN_LANGUAGES = frozenset({"", "other", "unknown"})
|
||||
_PERL_LINE = re.compile(r"^\s*(?:sub\s+\w+\s*[{(;]|use\s+strict\s*;|my\s+[$@%]\w+)")
|
||||
_RUBY_DEF = re.compile(r"^\s*def\s+(?:self\.)?\w+[?!=]?(?:\s*\(|\s|$)")
|
||||
_RUBY_END = re.compile(r"^\s*end\s*$")
|
||||
_BODY_SCAN_LINES = 40 # how far past a declaration head the body's ``{`` (or a prototype's ``;``) may be
|
||||
_STATEMENT_HEAD = re.compile(r"\s*(return|if|while|for|switch|else|case|throw|await|yield)\b")
|
||||
_FUNCTION_EXPR = r"(?:async\s+)?(?:function\b\s*\*?\s*[\w$]*\s*)?(?=\(|[\w$]+\s*=>)" # function (...) / (...) => / a =>
|
||||
|
||||
|
||||
def resolve_language(lines: Sequence[str], language: str) -> str:
|
||||
"""The alias-normalised language; a file given as unknown (``other``, the feature vocabulary's bucket for Go, Ruby,
|
||||
C#, Perl, ...) is sniffed for Perl (``sub name {``, ``use strict;``) and Ruby (``def`` ... ``end``), whose
|
||||
declaration forms a brace matcher never finds and whose ``#`` comments must be redacted."""
|
||||
language = LANGUAGE_ALIASES.get(language, language)
|
||||
if language in _UNKNOWN_LANGUAGES and lines:
|
||||
if any(_PERL_LINE.match(line) for line in lines):
|
||||
return "perl"
|
||||
if any(_RUBY_DEF.match(line) for line in lines) and any(_RUBY_END.match(line) for line in lines):
|
||||
return "ruby"
|
||||
return language
|
||||
|
||||
|
||||
def _blank_comments(lines: Sequence[str], language: str) -> list[str]:
|
||||
style = language if language in SLASH_COMMENTS or language in HASH_COMMENTS else "c"
|
||||
code: list[str] = []
|
||||
state: str | None = None
|
||||
for line in lines:
|
||||
ranges, state = _comment_ranges(line, state, style)
|
||||
for start, end in reversed(ranges):
|
||||
line = line[:start] + " " * (end - start) + line[end:]
|
||||
code.append(line)
|
||||
return code
|
||||
|
||||
|
||||
def _declaration_heads(name: str) -> re.Pattern[str]:
|
||||
"""Where a brace language may declare ``name`` (the body follows): ``name(`` (a definition -- or a call, told
|
||||
apart by :func:`_body_start`); Perl ``sub name {``; an object-literal method ``name: function (`` /
|
||||
``name: (a) =>``; an assignment ``x.y.name = function (`` / ``= (a) =>`` / ``const name = m.name = async a =>``
|
||||
(any chain). ``$`` is an identifier character (``Glance$serveRequest``)."""
|
||||
quoted = re.escape(name)
|
||||
return re.compile(
|
||||
rf"(?<![\w.>$]){quoted}\s*(?:<[^<>(){{}};]*>\s*)?(?=\()"
|
||||
rf"|^\s*sub\s+{quoted}\s*(?=[{{(;])"
|
||||
rf"|(?<![\w$]){quoted}\s*:\s*{_FUNCTION_EXPR}"
|
||||
rf"|(?<![\w$]){quoted}\s*=\s*(?:[\w$.\[\]'\"]+\s*=\s*)*{_FUNCTION_EXPR}"
|
||||
)
|
||||
|
||||
|
||||
def _body_start(code: Sequence[str], line: int, column: int) -> tuple[str, int, int] | None:
|
||||
"""Scan from a declaration head for its body: ``("brace", line, column)`` of the ``{`` at parenthesis depth 0,
|
||||
``("expression", line, column)`` of the ``;`` or ``,`` ending an arrow function's expression body, ``None`` for a
|
||||
prototype (``;`` before any body), a call (a ``)`` closing an outer parenthesis, a ``}``) or no body within
|
||||
:data:`_BODY_SCAN_LINES`. Parentheses are tracked so ``f(a, function () {})`` is not a definition of ``f``."""
|
||||
depth, arrow = 0, False
|
||||
for j in range(line, min(line + _BODY_SCAN_LINES, len(code))):
|
||||
text = code[j]
|
||||
k = column if j == line else 0
|
||||
while k < len(text):
|
||||
c = text[k]
|
||||
if c == "(":
|
||||
depth += 1
|
||||
elif c == ")":
|
||||
depth -= 1
|
||||
if depth < 0:
|
||||
return None
|
||||
elif depth == 0:
|
||||
if c == "{":
|
||||
return ("brace", j, k)
|
||||
if arrow and c in ";,":
|
||||
return ("expression", j, k)
|
||||
if c in "};": # a statement ends (a call, a prototype) before any body opened
|
||||
return None
|
||||
if text.startswith("=>", k):
|
||||
arrow = True
|
||||
k += 1
|
||||
k += 1
|
||||
return None
|
||||
|
||||
|
||||
def _brace_bounds(lines: Sequence[str], language: str, function: str) -> tuple[int, int] | None:
|
||||
"""A brace language's definition of ``function`` (``Class::name`` and ``a.b`` reduced to the last name): the first
|
||||
declaration head (:func:`_declaration_heads`) whose body :func:`_body_start` finds, to the matching ``}`` (or the
|
||||
end of an expression body). Comments are blanked before anything is matched or counted."""
|
||||
name = re.split(r"::|\.|->", function)[-1]
|
||||
if not re.fullmatch(r"[A-Za-z_$][\w$]*", name):
|
||||
return None
|
||||
heads = _declaration_heads(name)
|
||||
code = _blank_comments(lines, language)
|
||||
for i, line in enumerate(code):
|
||||
if _STATEMENT_HEAD.match(line):
|
||||
continue
|
||||
for found in heads.finditer(line):
|
||||
body = _body_start(code, i, found.end())
|
||||
if body is None:
|
||||
continue
|
||||
kind, j, k = body
|
||||
if kind == "expression":
|
||||
return i, j + 1
|
||||
depth = 0
|
||||
for row in range(j, len(code)):
|
||||
for c in code[row][k:] if row == j else code[row]:
|
||||
if c == "{":
|
||||
depth += 1
|
||||
elif c == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return i, row + 1
|
||||
return i, len(code)
|
||||
return None
|
||||
|
||||
|
||||
def function_bounds(lines: Sequence[str], language: str, function: str) -> tuple[int, int] | None:
|
||||
"""0-based ``[start, end)`` of the first declaration of ``function`` (the whole file for ``<global>``), or
|
||||
``None`` when the file does not declare it. Languages without a repo-facts declaration pattern (C, C++, C#, Perl,
|
||||
...) and brace languages whose pattern misses the name use the brace matcher; an unknown language is sniffed
|
||||
(:func:`resolve_language`)."""
|
||||
language = resolve_language(lines, language)
|
||||
if function == GLOBAL:
|
||||
return (0, len(lines)) if lines else None
|
||||
pattern = DECLARATION.get(language)
|
||||
start = next((i for i, line in enumerate(lines) if _declared_name(pattern, line) == function), None) if pattern else None
|
||||
if start is None:
|
||||
return _brace_bounds(lines, language, function) if language in SLASH_COMMENTS or pattern is None else None
|
||||
span = function_span(lines, language, function)
|
||||
if not span:
|
||||
return None
|
||||
return start, min(start + span, len(lines))
|
||||
|
||||
|
||||
def _window(start: int, end: int, focus: Sequence[int], size: int) -> tuple[int, int]:
|
||||
if end - start <= size:
|
||||
return start, end
|
||||
|
||||
@@ -20,9 +20,11 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from ..model.taxonomy import load_families
|
||||
from ..plane.tasks.repo_facts import _COMMENT_PREFIXES, DECLARATION, GLOBAL, _declared_name, enclosing, product_files
|
||||
from ..preprocess import build_file_target, detect_language
|
||||
from ..ruleset import PatternRule
|
||||
from ..source import _COMMENT_PREFIXES, DECLARATION, MODULE_FUNCTION, _declared_name, function_span
|
||||
from ..source import enclosing_at as enclosing
|
||||
from ..source import product_paths as product_files
|
||||
from .roles import CWE_OPERATION, SOURCE_KIND, VERSION, Function, RoleSet, body_roles, functions_of, origin_of
|
||||
from .schema import (
|
||||
BY_NAME,
|
||||
@@ -218,35 +220,15 @@ def engine_part(
|
||||
def facts_part(callers: Sequence[Mapping[str, Any]], function: str, version: str | None = None) -> Part:
|
||||
"""``callers`` are the repo-facts call sites of the candidate's function in other files."""
|
||||
files = {str(site.get("path")) for site in callers}
|
||||
return Part("ran", {"facts.callers": len(callers), "facts.caller_files": len(files), "facts.is_global": function == GLOBAL}, version)
|
||||
|
||||
|
||||
def function_span(lines: Sequence[str], language: str, function: str) -> int | None:
|
||||
"""Lines of the first declaration of ``function`` up to the next declaration (Python: up to the next line indented
|
||||
no deeper), ``None`` when the file does not declare it. ``<global>`` counts the whole file."""
|
||||
if function == GLOBAL:
|
||||
return len(lines)
|
||||
pattern = DECLARATION.get(language)
|
||||
if pattern is None:
|
||||
return None
|
||||
start = next((i for i, line in enumerate(lines) if _declared_name(pattern, line) == function), None)
|
||||
if start is None:
|
||||
return None
|
||||
indent = len(lines[start]) - len(lines[start].lstrip(" \t"))
|
||||
for end in range(start + 1, len(lines)):
|
||||
line = lines[end]
|
||||
if language == "python":
|
||||
if line.strip() and not line.lstrip().startswith("#") and len(line) - len(line.lstrip(" \t")) <= indent:
|
||||
return end - start
|
||||
elif _declared_name(pattern, line):
|
||||
return end - start
|
||||
return len(lines) - start
|
||||
return Part("ran", {"facts.callers": len(callers), "facts.caller_files": len(files), "facts.is_global": function == MODULE_FUNCTION},
|
||||
version,
|
||||
)
|
||||
|
||||
|
||||
def source_part(lines: Sequence[str] | None, language: str, function: str) -> Part:
|
||||
"""The function's span; ``none`` without the file or a declaration pattern for its language (C, say: no coverage,
|
||||
not a failure), ``failed`` when a covered file does not declare the function."""
|
||||
if lines is None or (function != GLOBAL and language not in DECLARATION):
|
||||
if lines is None or (function != MODULE_FUNCTION and language not in DECLARATION):
|
||||
return NONE
|
||||
span = function_span(lines, language, function)
|
||||
return Part("ran", {"facts.function_lines": span}) if span is not None else Part("failed")
|
||||
@@ -310,7 +292,7 @@ def model_sinks_part(roles: RoleSet | None, path: str, function: str, version: s
|
||||
|
||||
def function_of(lines: Sequence[str] | None, path: str, language: str, function: str) -> Function | None:
|
||||
"""The declared function ``function`` of a file (its parameters and body), ``None`` when it is not declared."""
|
||||
if not lines or function == GLOBAL:
|
||||
if not lines or function == MODULE_FUNCTION:
|
||||
return None
|
||||
return next((f for f in functions_of(path, lines, language) if f.name == function), None)
|
||||
|
||||
@@ -344,7 +326,7 @@ class SourceIndex:
|
||||
language = detect_language(Path(path))
|
||||
if lines and language in DECLARATION and line and 0 < line <= len(lines):
|
||||
return enclosing(lines, line - 1, language)
|
||||
return fallback or GLOBAL
|
||||
return fallback or MODULE_FUNCTION
|
||||
|
||||
def callers(self) -> dict[str, set[str]]:
|
||||
"""Function name -> the names of the functions that call it anywhere in the product files."""
|
||||
@@ -421,7 +403,7 @@ def _signals(
|
||||
|
||||
def _callers_of(index: SourceIndex, callers: Mapping[str, set[str]], path: str, function: str) -> list[dict[str, Any]]:
|
||||
"""The call sites of ``function`` in other product files, as repo-facts records them (capped at 40)."""
|
||||
if function == GLOBAL or function not in callers:
|
||||
if function == MODULE_FUNCTION or function not in callers:
|
||||
return []
|
||||
sites: list[dict[str, Any]] = []
|
||||
pattern = re.compile(r"(?<![A-Za-z0-9_$])" + re.escape(function) + r"\s*\(")
|
||||
|
||||
@@ -45,9 +45,10 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from ..model.taxonomy import load_families
|
||||
from ..plane.tasks.repo_facts import _EXCLUDED_DIRS, _NOT_PRODUCT, DECLARATION, GLOBAL, _declared_name, declared_functions, enclosing
|
||||
from ..preprocess import detect_language
|
||||
from ..ruleset.frameworks import load_frameworks
|
||||
from ..source import _EXCLUDED_DIRS, _NOT_PRODUCT, DECLARATION, MODULE_FUNCTION, _declared_name, declared_functions, module_function
|
||||
from ..source import enclosing_at as enclosing
|
||||
|
||||
DEFAULT_SOURCES = Path(__file__).resolve().parent / "sources.toml"
|
||||
DEFAULT_CACHE = Path.home() / ".cache" / "openultrasast"
|
||||
@@ -561,7 +562,9 @@ class LabelBuilder:
|
||||
vulnerable, fixed = str(case["vulnerable"]), str(case["fixed"])
|
||||
old, new = clone.ranges(vulnerable, fixed)
|
||||
if source.options.get("matching") == "declared_sites":
|
||||
positives = {(site.partition("::")[0], site.partition("::")[2]) for site in case.get("sites", []) if "::" in site}
|
||||
positives = {
|
||||
(site.partition("::")[0], module_function(site.partition("::")[2])) for site in case.get("sites", []) if "::" in site
|
||||
}
|
||||
else:
|
||||
positives = changed_functions(clone, vulnerable, old)
|
||||
fixed_changed = changed_functions(clone, fixed, new)
|
||||
@@ -670,12 +673,12 @@ class LabelBuilder:
|
||||
fix (a fix commit distinct from the vulnerable one): exactly one function the fixed side declares that the
|
||||
vulnerable side does not, with the labelled function's parameter list and a changed body (``fixed_side_moved``).
|
||||
A different function of the same snapshot (a guard helper, a sibling handler) is never a negative."""
|
||||
from .excerpt import LANGUAGE_ALIASES, function_bounds
|
||||
from ..source import canonical_language, function_bounds
|
||||
|
||||
fixed_lines, vuln_lines = self._pair_lines(case.fixed_file), self._pair_lines(case.vuln_file)
|
||||
if not fixed_lines:
|
||||
return FixedSide(ABSENT_UNREAD)
|
||||
language = LANGUAGE_ALIASES.get(case.language, case.language)
|
||||
language = canonical_language(case.language)
|
||||
if function_bounds(fixed_lines, language, function):
|
||||
return FixedSide(FIXED_DECLARED)
|
||||
vuln_span = function_bounds(vuln_lines or (), language, function)
|
||||
@@ -721,8 +724,9 @@ class LabelBuilder:
|
||||
if not known.get("in_scope", True) or not known.get("function") or not known.get("file"):
|
||||
self.build.skip("recipes: known entry without an in-scope function")
|
||||
continue
|
||||
candidate = f"{known['file']}::{module_function(str(known['function']))}"
|
||||
self._row(
|
||||
candidate=f"{known['file']}::{known['function']}", family=str(known.get("family", "unknown")), label=1,
|
||||
candidate=candidate, family=str(known.get("family", "unknown")), label=1,
|
||||
source=source.id,
|
||||
source_ref=f"{recipe.get('name')}:{known.get('id', '')}", unit="pin", weight=1.0, split=source.id, group=repo_name(url),
|
||||
repo=url, language=_language(str(known["file"])), evidence="reviewed known vulnerability", pin=commit,
|
||||
@@ -743,11 +747,13 @@ class LabelBuilder:
|
||||
self.build.skip("adjudications: verdict, case or site unreadable")
|
||||
continue
|
||||
population, case = found
|
||||
site_path, line, function = match.group("path"), int(match.group("line")), match.group("function").strip()
|
||||
site_path, line = match.group("path"), int(match.group("line"))
|
||||
function = module_function(match.group("function").strip())
|
||||
if not function:
|
||||
lines = self._clone(population, str(case["id"])).lines(str(case["vulnerable"]), site_path)
|
||||
language = detect_language(Path(site_path))
|
||||
function = enclosing(lines, line - 1, language) if lines and language in DECLARATION and line <= len(lines) else GLOBAL
|
||||
covered = lines and language in DECLARATION and line <= len(lines)
|
||||
function = enclosing(lines, line - 1, language) if covered else MODULE_FUNCTION
|
||||
self._row(
|
||||
candidate=f"{site_path}::{function}", family=str(row.get("family") or case["family"]), label=label, source=source.id,
|
||||
source_ref=f"{Path(path).name}#{key}[{index}]", unit="pin", weight=1.0, split=population.id,
|
||||
|
||||
@@ -31,10 +31,11 @@ from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from ..plane.tasks.repo_facts import DECLARATION, GLOBAL, KEYWORDS, _declared_name, product_files
|
||||
from ..preprocess import detect_language
|
||||
from ..ruleset.frameworks import Priors, normalize_priors, tag_of
|
||||
from ..semantic.facts import _LANGUAGE_ALIASES, FactLoadError, SanitizerFact, SemanticFacts, SinkFact, SourceFact, load_facts, with_priors
|
||||
from ..semantic.facts import FactLoadError, SanitizerFact, SemanticFacts, SinkFact, SourceFact, load_facts, with_priors
|
||||
from ..source import DECLARATION, KEYWORDS, MODULE_FUNCTION, _declared_name, canonical_language
|
||||
from ..source import product_paths as product_files
|
||||
from .schema import OPERATIONS, ORIGINS, SOURCE_KINDS
|
||||
|
||||
ROLE_KINDS = ("sink", "source", "sanitizer")
|
||||
@@ -128,7 +129,7 @@ class RoleSet:
|
||||
|
||||
def language_key(language: str) -> str:
|
||||
"""The semantic-facts language of a detected language (TypeScript shares JavaScript's entries)."""
|
||||
return _LANGUAGE_ALIASES.get(language, language)
|
||||
return canonical_language(language)
|
||||
|
||||
|
||||
def _facts(language: str, priors: Priors, facts: SemanticFacts | None) -> SemanticFacts:
|
||||
@@ -262,7 +263,7 @@ def functions_of(path: str, lines: Sequence[str], language: str, *, dependency:
|
||||
return []
|
||||
starts: dict[str, int] = {}
|
||||
bodies: dict[str, list[str]] = {}
|
||||
current = GLOBAL
|
||||
current = MODULE_FUNCTION
|
||||
stack: list[tuple[int, str]] = [] # Python: the open (indent, name) declarations, innermost last
|
||||
for index, line in enumerate(lines):
|
||||
declared = _declared_name(pattern, line)
|
||||
@@ -276,11 +277,11 @@ def functions_of(path: str, lines: Sequence[str], language: str, *, dependency:
|
||||
stack.pop()
|
||||
if declared:
|
||||
stack.append((indent, declared))
|
||||
owner = stack[-1][1] if stack else GLOBAL
|
||||
owner = stack[-1][1] if stack else MODULE_FUNCTION
|
||||
else:
|
||||
current = declared or current
|
||||
owner = current
|
||||
if owner != GLOBAL:
|
||||
if owner != MODULE_FUNCTION:
|
||||
bodies.setdefault(owner, []).append(line)
|
||||
out = []
|
||||
for name, start in sorted(starts.items(), key=lambda item: item[1]):
|
||||
@@ -450,7 +451,7 @@ def from_model_roles(payload: Mapping[str, Any]) -> RoleSet:
|
||||
roles.append(
|
||||
Role(
|
||||
detect_language(Path(path)), kind, name or call, operation if operation in vocabulary else "other", "model_role",
|
||||
0, "", path, str(row.get("function") or GLOBAL),
|
||||
0, "", path, str(row.get("function") or MODULE_FUNCTION),
|
||||
)
|
||||
) # fmt: skip
|
||||
return RoleSet(tuple(sorted(set(roles))), tuple(sorted(set(payload.get("unclassified_paths") or ()))), ("model_role",))
|
||||
|
||||
@@ -20,6 +20,8 @@ from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Literal
|
||||
|
||||
from ..source import canonical_language
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .taxonomy import FamilyTaxonomy
|
||||
|
||||
@@ -46,7 +48,6 @@ FAMILY_SHAPE: dict[str, tuple[CandidateKind, ...]] = {
|
||||
# The p90 of the measured distribution is 22 sites per labeled function, so two batches cover all but the tail.
|
||||
MAX_CANDIDATES_PER_CALL = 12
|
||||
_MAX_TEXT = 400
|
||||
_LANGUAGE = {"python": "python", "javascript": "javascript", "typescript": "typescript", "c_cpp": "c", "java": "java"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -233,7 +234,7 @@ def _line_text(lines: Sequence[str], line: int) -> str:
|
||||
def _language_of(path: str) -> str:
|
||||
from ..preprocess import detect_language
|
||||
|
||||
return _LANGUAGE.get(detect_language(Path(path)), detect_language(Path(path)))
|
||||
return canonical_language(detect_language(Path(path)))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
||||
+13
-13
@@ -40,8 +40,9 @@ from .rank import rank_targets
|
||||
from .ruleset import PatternRule
|
||||
from .semantic import OverlayRecord, adjudicate
|
||||
from .semantic.facts import FactLoadError, load_facts
|
||||
from .semantic.functions import RESERVED_FUNCTION_LABELS, function_at, named_function_ranges, spans_named
|
||||
from .semantic.obligations.facts import OPERATION_KINDS
|
||||
from .source import MODULE_FUNCTION, RESERVED_FUNCTION_LABELS, FunctionRange, enclosing, spans_named
|
||||
from .source import ranges as source_ranges
|
||||
|
||||
DEFAULT_CATALOG = Path("benchmarks/pairs/catalog.toml")
|
||||
DEFAULT_SAST_CATALOG = Path("benchmarks/pairs/sast/catalog.toml")
|
||||
@@ -57,7 +58,7 @@ DEFAULT_MECHANISMS = Path("benchmarks/pairs/mechanisms.toml")
|
||||
OVERLAY_SLICES = frozenset({"sast", "vfc", "vibe-py", "vfc-js", "agent-vfc", "owasp"})
|
||||
# Harvested function slices: every label must name the function (design §LabelSchema); "<anon>" does not scope.
|
||||
FUNCTION_REQUIRED_SLICES = OVERLAY_SLICES - {"sast"}
|
||||
FunctionRanges = dict[str, tuple[tuple[str, int, int], ...]]
|
||||
FunctionRanges = dict[str, tuple[FunctionRange, ...]]
|
||||
SLICE_NAMES = ("all", "local", "github", "sast", "vfc", "vibe-py", "vfc-js", "agent-vfc", "owasp")
|
||||
PROVENANCES = frozenset({"human", "agent", "mixed", "synthetic"})
|
||||
SPLITS = frozenset({"train", "holdout"})
|
||||
@@ -806,7 +807,7 @@ class _Side:
|
||||
findings: list[StaticFinding]
|
||||
records: list[OverlayRecord]
|
||||
parse_failed: int
|
||||
ranges: dict[str, tuple[tuple[str, int, int], ...]]
|
||||
ranges: dict[str, tuple[FunctionRange, ...]]
|
||||
obligations: list[StaticFinding] = field(default_factory=list) # obligation statics, computed for obligation-labeled rows only
|
||||
|
||||
|
||||
@@ -815,18 +816,17 @@ def _overlay_scan(target: Path, ruleset: tuple[PatternRule, ...] | None = None,
|
||||
findings = [f for f in quick_scan_findings(target, targets, rank_targets(targets), ruleset) if f.status != "shadow"]
|
||||
records = adjudicate(root=target, targets=targets, findings=findings)
|
||||
parse_failed = 0
|
||||
ranges: dict[str, tuple[tuple[str, int, int], ...]] = {}
|
||||
ranges: dict[str, tuple[FunctionRange, ...]] = {}
|
||||
for item in targets:
|
||||
try:
|
||||
text = (target / item.path).read_text(errors="ignore")
|
||||
except OSError:
|
||||
parse_failed += 1
|
||||
continue
|
||||
named = named_function_ranges(item.path, text, item.language)
|
||||
if named is None:
|
||||
try:
|
||||
ranges[item.path] = source_ranges(item.path, text, item.language)
|
||||
except ValueError:
|
||||
parse_failed += 1
|
||||
else:
|
||||
ranges[item.path] = named
|
||||
statics = _obligation_scan(target, targets) if obligations else []
|
||||
return _Side(findings=findings, records=records, parse_failed=parse_failed, ranges=ranges, obligations=statics)
|
||||
|
||||
@@ -890,19 +890,19 @@ def _with_functions(root: Path, findings: list[StaticFinding]) -> list[StaticFin
|
||||
"""Give regex findings an enclosing function from a parse so function-labeled rows can match (Req 1.2)."""
|
||||
if not any(finding.function_name is None for finding in findings):
|
||||
return findings
|
||||
ranges: dict[str, tuple[tuple[str, int, int], ...]] = {}
|
||||
ranges: dict[str, tuple[FunctionRange, ...]] = {}
|
||||
for target in _targets(root):
|
||||
try:
|
||||
named = named_function_ranges(target.path, (root / target.path).read_text(errors="ignore"), target.language)
|
||||
except OSError:
|
||||
named = source_ranges(target.path, (root / target.path).read_text(errors="ignore"), target.language)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
if named is not None:
|
||||
ranges[target.path] = named
|
||||
out: list[StaticFinding] = []
|
||||
for finding in findings:
|
||||
if finding.function_name is None:
|
||||
function = function_at(ranges.get(finding.path, ()), finding.line)
|
||||
if function is not None:
|
||||
function = enclosing(ranges.get(finding.path, ()), finding.line)
|
||||
if function != MODULE_FUNCTION:
|
||||
finding = replace(finding, function_name=function)
|
||||
out.append(finding)
|
||||
return out
|
||||
|
||||
@@ -40,10 +40,10 @@ from typing import Any
|
||||
import yaml
|
||||
|
||||
from ..preprocess import detect_language
|
||||
from ..source import DECLARATION, MODULE_FUNCTION, _declared_name
|
||||
from .generate import ATESPACE, CALLS_PER_HUNT, CASE_PATH, INPUTS_PATH, _dump, _json, _task
|
||||
from .manifests import GIT_COMMITS_ANNOTATION, RUN_API_VERSION, Task
|
||||
from .reconciler import _ax_name, _doc
|
||||
from .tasks.repo_facts import DECLARATION, GLOBAL, _declared_name
|
||||
from .tasks.roles import CHUNK_LINES
|
||||
from .tasks.verify import OPERATIONS, units_of
|
||||
|
||||
@@ -106,9 +106,9 @@ def blob_sha1(text: str) -> str:
|
||||
|
||||
|
||||
def declaration_line(lines: Sequence[str] | None, path: str, function: str) -> int:
|
||||
"""1-based line of the first declaration of ``function`` in ``lines``; 1 for ``<global>`` or when not declared."""
|
||||
"""1-based line of the first declaration of ``function`` in ``lines``; 1 for ``<module>`` or when not declared."""
|
||||
pattern = DECLARATION.get(detect_language(Path(path)) or "")
|
||||
if not lines or pattern is None or function == GLOBAL:
|
||||
if not lines or pattern is None or function == MODULE_FUNCTION:
|
||||
return 1
|
||||
return next((i + 1 for i, line in enumerate(lines) if _declared_name(pattern, line) == function), 1)
|
||||
|
||||
@@ -116,7 +116,7 @@ def declaration_line(lines: Sequence[str] | None, path: str, function: str) -> i
|
||||
def declares(lines: Sequence[str] | None, path: str, function: str) -> bool:
|
||||
"""Whether a side's file declares ``function``, by the excerpt builder's matcher
|
||||
(:func:`..learn.excerpt.function_bounds`): a candidate is asked only where an excerpt of it exists."""
|
||||
from ..learn.excerpt import function_bounds
|
||||
from ..source import function_bounds
|
||||
|
||||
return bool(lines) and function_bounds(lines or (), detect_language(Path(path)) or "", function) is not None
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ what ``ousast scan --mode quick`` runs -- on the case's vulnerable pin and on it
|
||||
rule_id, rule_status (enabled | shadow), path, line, function, pin_role (vulnerable | fixed), pin, in_fix_range
|
||||
|
||||
``function`` is the enclosing declaration by the patterns ``repo-facts`` uses (so an alert joins the verify
|
||||
candidates' ``path::function``; ``<global>`` outside any function); a language those patterns do not cover keeps the
|
||||
candidates' ``path::function``; ``<module>`` outside any function); a language those patterns do not cover keeps the
|
||||
scan's own reachability attribution. ``in_fix_range`` says whether the line lies inside the fix: the old-side ranges
|
||||
of ``case.json`` on the vulnerable pin, the new-side ``fixed_ranges`` on the fixed pin (null when ``case.json`` has
|
||||
none). A fixed-pin alert inside the fix is a false alert on code the fix wrote.
|
||||
@@ -43,7 +43,8 @@ from ...mapping import analyze_entry_points, attach_reachability_hints
|
||||
from ...preprocess import detect_language, preprocess_repository
|
||||
from ...rank import rank_targets
|
||||
from ...ruleset import DEFAULT_RULESET_DIR, load_ruleset
|
||||
from .repo_facts import DECLARATION, GLOBAL, enclosing
|
||||
from ...source import DECLARATION, MODULE_FUNCTION
|
||||
from ...source import enclosing_at as enclosing
|
||||
|
||||
ROLES = ("vulnerable", "fixed")
|
||||
DIRS = ("OUSAST_WORKSPACE_DIR", "OUSAST_FIXED_DIR")
|
||||
@@ -105,12 +106,12 @@ def in_range(ranges: Ranges | None, path: str, line: int | None) -> bool | None:
|
||||
|
||||
|
||||
def function_at(root: Path, path: str, line: int | None, fallback: str | None, cache: dict[str, list[str]]) -> str:
|
||||
"""The enclosing declaration of ``path:line`` by the ``repo-facts`` patterns, else ``fallback`` (``<global>``)."""
|
||||
"""The enclosing declaration of ``path:line`` by the ``repo-facts`` patterns, else ``fallback`` (``<module>``)."""
|
||||
lines = cache.setdefault(path, _lines(root / path))
|
||||
language = detect_language(root / path)
|
||||
if language in DECLARATION and line and 0 < line <= len(lines):
|
||||
return enclosing(lines, line - 1, language)
|
||||
return fallback or GLOBAL
|
||||
return fallback or MODULE_FUNCTION
|
||||
|
||||
|
||||
def scan(root: Path, role: str, pin: str, ranges: Ranges | None) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||||
|
||||
@@ -55,8 +55,8 @@ from ...learn.roles import from_model_roles, infer_for_checkout
|
||||
from ...model.taxonomy import load_families
|
||||
from ...preprocess import detect_language
|
||||
from ...ruleset import DEFAULT_RULESET_DIR, PatternRule, load_ruleset
|
||||
from ...source import MODULE_FUNCTION
|
||||
from .alerts import covers
|
||||
from .repo_facts import GLOBAL
|
||||
|
||||
OUTPUT = "features.jsonl"
|
||||
PASSES = ("a", "b", "c")
|
||||
@@ -142,7 +142,7 @@ def case_features(
|
||||
rule_id = str(alert.get("rule_id") or "")
|
||||
engine = rule_id.startswith(ENGINE_PREFIX)
|
||||
family = rule_id.removeprefix(ENGINE_PREFIX) if engine else _family(by_id.get(rule_id))
|
||||
key = (f"{alert.get('path')}::{alert.get('function') or GLOBAL}", family)
|
||||
key = (f"{alert.get('path')}::{alert.get('function') or MODULE_FUNCTION}", family)
|
||||
entry = candidates.setdefault(key, {"line": alert.get("line"), "quick": [], "engine": []})
|
||||
if engine:
|
||||
entry["engine"].append({"site": f"{alert.get('path')}:{alert.get('line')}"})
|
||||
@@ -179,7 +179,7 @@ def case_features(
|
||||
entries = entry_names(index.root, {c.partition("::")[0] for c, _ in candidates})
|
||||
parts["source"] = source_part(index.lines(path), language, function)
|
||||
parts["roles"] = roles_part(function_of(index.lines(path), path, language, function), inferred, language)
|
||||
distance = 0 if function == GLOBAL and path in entries[1] else entry_distance(function, entries[0], index.callers())
|
||||
distance = 0 if function == MODULE_FUNCTION and path in entries[1] else entry_distance(function, entries[0], index.callers())
|
||||
parts["entry_points"] = Part("ran", {"facts.entry_distance": distance})
|
||||
out.append(record(candidate, family, language, "plane", parts))
|
||||
return out
|
||||
|
||||
@@ -13,172 +13,39 @@ declare them.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import traceback
|
||||
from collections.abc import Iterable, Mapping, Sequence
|
||||
from collections.abc import Mapping, Sequence
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from ...model.layout import is_test_path, is_vendored, layout_facts
|
||||
from ...preprocess import IGNORED_DIRS, detect_language
|
||||
from ...preprocess import detect_language
|
||||
from ...source import ( # noqa: F401 (re-exported: generate, remember and the tests import them from here)
|
||||
DECLARATION,
|
||||
KEYWORDS,
|
||||
MODULE_FUNCTION,
|
||||
_declared_name,
|
||||
_digest,
|
||||
declared_functions,
|
||||
)
|
||||
from ...source import call_sites_in as call_sites
|
||||
from ...source import enclosing_at as enclosing
|
||||
from ...source import product_paths as product_files
|
||||
|
||||
CALLERS_CAP = 40
|
||||
MAX_FILE_BYTES = 1_000_000
|
||||
LANGUAGES = frozenset({"python", "php", "javascript", "typescript", "java", "go", "ruby"})
|
||||
GLOBAL = "<global>"
|
||||
|
||||
_EXCLUDED_DIRS = frozenset(
|
||||
IGNORED_DIRS | {"vendor", "node_modules", "bower_components", "third_party", "third-party", "dist", "build", "out", "target"}
|
||||
)
|
||||
_NOT_PRODUCT = re.compile(
|
||||
r"(^|/)(tests?|__tests__|__mocks__|spec|specs|docs?|examples?|fixtures?|e2e|cypress|storybook|testdata)/"
|
||||
r"|\.(test|spec|stories|mock|e2e)\.[cm]?[jt]sx?$"
|
||||
r"|_test\.(py|go|rb)$|(^|/)test_[^/]*\.(py|rb)$|Test\.java$|Tests\.java$|_spec\.rb$"
|
||||
r"|\.min\.(js|css)$|\.bundle\.js$"
|
||||
)
|
||||
|
||||
_JS_MODIFIERS = r"(?:(?:export|default|public|private|protected|static|async|readonly|override)\s+)*"
|
||||
_JAVA_MODIFIERS = r"(?:(?:public|private|protected|static|final|abstract|synchronized|native|default|strictfp)\s+)*"
|
||||
DECLARATION: dict[str, re.Pattern[str]] = {
|
||||
"python": re.compile(r"^\s*(?:async\s+)?def\s+(\w+)\s*\("),
|
||||
"php": re.compile(r"^\s*(?:(?:public|private|protected|static|final|abstract)\s+)*function\s+&?(\w+)\s*\("),
|
||||
"javascript": re.compile(
|
||||
r"^\s*" + _JS_MODIFIERS + r"function\s*\*?\s*(\w+)\s*\("
|
||||
r"|^\s*" + _JS_MODIFIERS + r"(\w+)\s*\([^;]*\)\s*(?::[^{=]+)?\{\s*$"
|
||||
r"|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:function\b|\([^)]*\)\s*(?::[^=]+)?=>|\w+\s*=>)"
|
||||
),
|
||||
"java": re.compile(
|
||||
r"^\s*" + _JAVA_MODIFIERS + r"(?:<[^>]*>\s+)?(?!return\b|new\b|throw\b)[\w<>\[\],.?]+\s+(\w+)\s*\([^;]*\)\s*"
|
||||
r"(?:throws\s+[\w.,\s]+)?\{?\s*$"
|
||||
),
|
||||
"go": re.compile(r"^\s*func\s+(?:\([^)]*\)\s*)?(\w+)\s*\("),
|
||||
"ruby": re.compile(r"^\s*def\s+(?:self\.)?(\w+[?!=]?)"),
|
||||
}
|
||||
DECLARATION["typescript"] = DECLARATION["javascript"]
|
||||
KEYWORDS = frozenset(
|
||||
{
|
||||
"if",
|
||||
"for",
|
||||
"while",
|
||||
"switch",
|
||||
"catch",
|
||||
"return",
|
||||
"function",
|
||||
"else",
|
||||
"do",
|
||||
"try",
|
||||
"with",
|
||||
"elif",
|
||||
"except",
|
||||
"new",
|
||||
"throw",
|
||||
"super",
|
||||
"constructor",
|
||||
"foreach",
|
||||
"match",
|
||||
"case",
|
||||
"unless",
|
||||
"until",
|
||||
}
|
||||
)
|
||||
_COMMENT_PREFIXES = ("#", "//", "/*", "*", "*/", "--")
|
||||
|
||||
|
||||
class TaskFailed(RuntimeError):
|
||||
"""A crash inside `run` that has already been recorded in `summary.json` (status `failed`, traceback)."""
|
||||
|
||||
|
||||
def product_files(workspace: Path) -> list[Path]:
|
||||
"""Product source files of the supported languages, sorted; tests, vendored trees, build output, minified
|
||||
bundles and dot-directories are not product (the classification of `model_sinks.py` and the layout facts)."""
|
||||
facts = layout_facts()
|
||||
files: list[Path] = []
|
||||
for path in sorted(workspace.rglob("*")):
|
||||
if path.is_symlink() or not path.is_file():
|
||||
continue
|
||||
relative = path.relative_to(workspace)
|
||||
parts = relative.parts
|
||||
if any(part.startswith(".") or part in _EXCLUDED_DIRS for part in parts[:-1]) or parts[-1].startswith("."):
|
||||
continue
|
||||
posix = relative.as_posix()
|
||||
if _NOT_PRODUCT.search(posix) or is_vendored(posix, facts) or is_test_path(posix, facts):
|
||||
continue
|
||||
if detect_language(path) not in LANGUAGES or path.stat().st_size > MAX_FILE_BYTES:
|
||||
continue
|
||||
files.append(path)
|
||||
return files
|
||||
|
||||
|
||||
def declared_functions(lines: Sequence[str], language: str) -> list[str]:
|
||||
"""The function names a file declares, in source order, each once."""
|
||||
pattern = DECLARATION.get(language)
|
||||
if pattern is None:
|
||||
return []
|
||||
names: list[str] = []
|
||||
for line in lines:
|
||||
name = _declared_name(pattern, line)
|
||||
if name and name not in names:
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
|
||||
def enclosing(lines: Sequence[str], index: int, language: str) -> str:
|
||||
"""The nearest declaration at or above line ``index`` (0-based), `<global>` when there is none."""
|
||||
pattern = DECLARATION.get(language)
|
||||
if pattern is None:
|
||||
return GLOBAL
|
||||
indent = _indent(lines[index]) if language == "python" else None
|
||||
for back in range(index, -1, -1):
|
||||
name = _declared_name(pattern, lines[back])
|
||||
if not name:
|
||||
continue
|
||||
if indent is None or back == index or _indent(lines[back]) < indent:
|
||||
return name
|
||||
indent = min(indent, _indent(lines[back])) # a sibling or deeper def: keep looking for the one that encloses
|
||||
return GLOBAL
|
||||
|
||||
|
||||
def _indent(line: str) -> int:
|
||||
return len(line) - len(line.lstrip(" \t"))
|
||||
|
||||
|
||||
def call_sites(lines: Sequence[str], language: str, names: Iterable[str]) -> list[dict[str, Any]]:
|
||||
"""Calls to any of ``names`` in one file: `{"function", "line", "enclosing"}`, one per name and line, in
|
||||
line order. A declaration line is not a call of the function it declares; comment lines are skipped."""
|
||||
wanted = sorted(set(names), key=lambda n: (-len(n), n))
|
||||
if not wanted:
|
||||
return []
|
||||
token = re.compile(r"(?<![A-Za-z0-9_$])(" + "|".join(re.escape(n) for n in wanted) + r")\s*\(")
|
||||
pattern = DECLARATION.get(language)
|
||||
found: list[dict[str, Any]] = []
|
||||
for index, line in enumerate(lines):
|
||||
if line.lstrip().startswith(_COMMENT_PREFIXES):
|
||||
continue
|
||||
declared = _declared_name(pattern, line) if pattern is not None else ""
|
||||
seen: set[str] = set()
|
||||
for match in token.finditer(line):
|
||||
name = match.group(1)
|
||||
if name == declared or name in seen:
|
||||
continue
|
||||
seen.add(name)
|
||||
found.append({"function": name, "line": index + 1, "enclosing": enclosing(lines, index, language)})
|
||||
return found
|
||||
|
||||
|
||||
def _declared_name(pattern: re.Pattern[str], line: str) -> str:
|
||||
match = pattern.match(line)
|
||||
if not match:
|
||||
return ""
|
||||
name = next((g for g in match.groups() if g), "")
|
||||
return "" if name in KEYWORDS else name
|
||||
|
||||
|
||||
def _read_lines(path: Path) -> list[str]:
|
||||
return path.read_text(encoding="utf-8", errors="replace").splitlines()
|
||||
|
||||
@@ -187,10 +54,6 @@ def _candidate_names(candidates: Sequence[Mapping[str, Any]]) -> list[str]:
|
||||
return sorted({str(c["function"]) for c in candidates if c.get("function")})
|
||||
|
||||
|
||||
def _digest(names: Sequence[str]) -> str:
|
||||
return hashlib.sha256("\n".join(names).encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
def _load_units(units_path: Path, digest: str) -> dict[str, dict[str, Any]]:
|
||||
"""Rows already in `units.jsonl` that were computed for the same candidate set (Requirement 2.2)."""
|
||||
rows: dict[str, dict[str, Any]] = {}
|
||||
@@ -240,7 +103,7 @@ def assemble_facts(
|
||||
for path, row in sorted(rows.items()):
|
||||
for call in row.get("calls", ()):
|
||||
by_name.setdefault(str(call["function"]), []).append(
|
||||
{"path": path, "line": int(call["line"]), "enclosing": str(call.get("enclosing", GLOBAL))}
|
||||
{"path": path, "line": int(call["line"]), "enclosing": str(call.get("enclosing", MODULE_FUNCTION))}
|
||||
)
|
||||
callers: dict[str, list[dict[str, Any]]] = {}
|
||||
for candidate in sorted(candidates, key=lambda c: (str(c["path"]), str(c["function"]))):
|
||||
|
||||
@@ -32,9 +32,9 @@ from typing import Any
|
||||
|
||||
from ...learn.roles import MODEL_ROLE_KINDS
|
||||
from ...learn.schema import OPERATIONS, SOURCE_KINDS
|
||||
from ...source import MODULE_FUNCTION, product_files
|
||||
from ...tool_hunter import ChatClient
|
||||
from ..budget import AccountError, BudgetExhausted, MeteredClient
|
||||
from .repo_facts import GLOBAL, product_files
|
||||
from .verify import build_client
|
||||
|
||||
CHUNK_LINES = 400
|
||||
@@ -52,7 +52,7 @@ PROMPT = (
|
||||
"- guard: checks a value and rejects it (allow-list, permission or ownership check);\n"
|
||||
"- dispatch: registers or invokes a callback through a string-keyed registry (hooks, events, signals).\n"
|
||||
"Answer with JSON only: "
|
||||
'{{"roles": [{{"role": "<sink|source|sanitizer|guard|dispatch>", "function": "<enclosing function, or <global>>", '
|
||||
'{{"roles": [{{"role": "<sink|source|sanitizer|guard|dispatch>", "function": "<enclosing function, or <module>>", '
|
||||
'"line": <line number>, "call": "<the call or expression>", "operation": "<the kind>", '
|
||||
'"evidence": "<the line, quoted>"}}]}}. Return {{"roles": []}} if there is none.\n\nFile: {path}\n\n{code}'
|
||||
)
|
||||
@@ -80,7 +80,7 @@ def parse_roles(content: str | None, path: str) -> list[dict[str, Any]]:
|
||||
vocabulary = SOURCE_KINDS if item["role"] == "source" else OPERATIONS
|
||||
operation = item.get("operation") if item.get("operation") in vocabulary else "other"
|
||||
named = item.get("function")
|
||||
function = named.strip() if isinstance(named, str) and named.strip() else GLOBAL
|
||||
function = named.strip() if isinstance(named, str) and named.strip() else MODULE_FUNCTION
|
||||
line = item["line"] if isinstance(item.get("line"), int) else 0
|
||||
out.append(
|
||||
{
|
||||
@@ -154,7 +154,7 @@ def run(
|
||||
log = output_dir / "units.jsonl"
|
||||
metered = budget if budget is not None else MeteredClient(client)
|
||||
model = model or "unbound"
|
||||
wanted = sorted(set(files)) if files is not None else [p.relative_to(workspace).as_posix() for p in product_files(workspace)]
|
||||
wanted = sorted(set(files)) if files is not None else product_files(workspace)
|
||||
missing = [p for p in wanted if not (workspace / p).is_file()]
|
||||
done = {p: row for p, row in _read_units(log).items() if p in wanted}
|
||||
status, reason = ("failed", f"files not in the workspace: {missing[:5]}") if missing else ("done", "")
|
||||
|
||||
@@ -12,19 +12,10 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from ..ruleset.frameworks import Priors, kept, normalize_priors, read_tag, tag_of
|
||||
from ..source import canonical_language
|
||||
|
||||
DEFAULT_FACTS_DIR = Path(__file__).resolve().parents[1] / "ruleset" / "semantic"
|
||||
|
||||
_LANGUAGE_ALIASES = {
|
||||
"c": "c",
|
||||
"cpp": "c",
|
||||
"python": "python",
|
||||
"javascript": "javascript",
|
||||
"typescript": "javascript",
|
||||
"java": "java",
|
||||
}
|
||||
|
||||
|
||||
class FactLoadError(ValueError):
|
||||
"""Raised when semantic fact files are missing or invalid."""
|
||||
|
||||
@@ -146,7 +137,11 @@ class SemanticFacts:
|
||||
layouts: tuple[LayoutFact, ...] = ()
|
||||
|
||||
def for_language(self, language: str) -> SemanticFacts:
|
||||
key = _LANGUAGE_ALIASES.get(language, language)
|
||||
key = canonical_language(language)
|
||||
if key == "cpp":
|
||||
key = "c"
|
||||
elif key == "typescript":
|
||||
key = "javascript"
|
||||
return SemanticFacts(
|
||||
version=self.version,
|
||||
sources=tuple(item for item in self.sources if item.language == key),
|
||||
|
||||
@@ -1,151 +1,10 @@
|
||||
"""Named function ranges shared by the scorer, the overlay, and the catalog generator.
|
||||
"""Function ranges now belong to source."""
|
||||
|
||||
Tree-sitter names an arrow function after its first bare parameter and gives
|
||||
object or class methods, `exports.x = function`, and IIFEs ``<anon>``. Labels in
|
||||
the pair catalogs name the declarator (``const h = ...``, ``exports.h = ...``,
|
||||
``h(req, res) {``). Every consumer must therefore name ranges the same way, or
|
||||
function-scoped matching silently fails (pair-corpus-honesty Req 1.2, design
|
||||
§Scorer). This module is the single source of that naming. Stdlib plus ``.ir``
|
||||
only; never imports ``overlay`` or ``pairs``.
|
||||
"""
|
||||
from ..source import FunctionRange, enclosing, ranges
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
|
||||
from .ir import FileIR, parse_file
|
||||
|
||||
FunctionRange = tuple[str, int, int]
|
||||
MODULE_FUNCTION = "<module>"
|
||||
ANON_FUNCTION = "<anon>"
|
||||
UNNAMED_FUNCTIONS = frozenset({MODULE_FUNCTION, ANON_FUNCTION})
|
||||
# Keywords a parser or a regex fallback can mistake for a function name; never valid labels.
|
||||
RESERVED_FUNCTION_LABELS = frozenset(
|
||||
{
|
||||
"function",
|
||||
"def",
|
||||
"return",
|
||||
"if",
|
||||
"else",
|
||||
"for",
|
||||
"while",
|
||||
"switch",
|
||||
"catch",
|
||||
"class",
|
||||
"async",
|
||||
"await",
|
||||
"lambda",
|
||||
"const",
|
||||
"let",
|
||||
"var",
|
||||
"export",
|
||||
"import",
|
||||
"module",
|
||||
"exports",
|
||||
"new",
|
||||
"this",
|
||||
"self",
|
||||
"anonymous",
|
||||
ANON_FUNCTION,
|
||||
}
|
||||
)
|
||||
|
||||
_IDENT = r"[A-Za-z_$][\w$]*"
|
||||
_ARROW_TAIL = r"(?:async\s*)?(?:\([^)]*\)|" + _IDENT + r")\s*=>"
|
||||
_DECL = re.compile(
|
||||
# def name( / function name( / export async function name(
|
||||
r"^\s*(?:export\s+)?(?:async\s+)?(?:def|function|sub)\s+(" + _IDENT + r")\s*[\({]|"
|
||||
# const name = [a.b.c =]* [async] (args) => | ident =>
|
||||
r"^\s*(?:export\s+)?(?:const|let|var)\s+(" + _IDENT + r")\s*=\s*(?:" + _IDENT + r"(?:\." + _IDENT + r")*\s*=\s*)*" + _ARROW_TAIL + r"|"
|
||||
# exports.name = / module.exports.name = [async] function | (args) =>
|
||||
r"^\s*(?:module\.)?exports\.(" + _IDENT + r")\s*=\s*(?:async\s*)?(?:function\b|" + _ARROW_TAIL + r")|"
|
||||
# object property: name: [async] function | (args) =>
|
||||
r"^\s*(?:async\s+)?(" + _IDENT + r")\s*:\s*(?:async\s*)?(?:function\b|" + _ARROW_TAIL + r")|"
|
||||
# C / Java / JS method or shorthand: [type] name(args) { (no call: the arg list carries no callback or literal)
|
||||
r"^\s*(?:[A-Za-z_][\w:<>\*\s,]*[\s\*])?(" + r"[A-Za-z_][\w:$]*" + r")\s*\(([^;{]*)\)\s*(?:const\s*)?\{?\s*$"
|
||||
)
|
||||
_CALLBACK_ARGS = re.compile(r"\bfunction\b|=>|['\"`]|^\s*\d")
|
||||
# A statement keyword before `name(` means the line calls or controls, it does not declare.
|
||||
_STATEMENT_PREFIX = re.compile(
|
||||
r"^\s*(?:return|print|yield|throw|await|new|else|case|delete|typeof|synchronized|with|elif|except|try|if|for|while|switch|catch|do)\b"
|
||||
)
|
||||
# Only these grammars name an arrow function after its bare parameter; elsewhere the parser name is trustworthy.
|
||||
_PARAM_NAMED_ARROWS = frozenset({"javascript", "typescript"})
|
||||
__all__ = ["FunctionRange", "enclosing", "ranges", "function_at"]
|
||||
|
||||
|
||||
def declarator_name(line: str) -> str | None:
|
||||
"""Function name declared on ``line``, or None when the line declares nothing (a call is not a declaration)."""
|
||||
match = _DECL.match(line)
|
||||
if not match:
|
||||
return None
|
||||
groups = match.groups()
|
||||
name = next((group for group in groups[:5] if group), None)
|
||||
if name is None:
|
||||
return None
|
||||
if groups[4] is not None:
|
||||
args = groups[5] or ""
|
||||
if _CALLBACK_ARGS.search(args):
|
||||
return None # `exec('...', function (err) {` is a call, not a declaration
|
||||
if _STATEMENT_PREFIX.match(line) or name in {"if", "for", "while", "switch", "catch", "return", "elif", "with", "synchronized"}:
|
||||
return None
|
||||
name = name.rsplit("::", 1)[-1] # `bool K::m(` declares `m`; catalog labels are unqualified
|
||||
return name if name not in RESERVED_FUNCTION_LABELS else None
|
||||
|
||||
|
||||
def named_ranges_from_ir(ir: FileIR, text: str) -> tuple[FunctionRange, ...]:
|
||||
"""Declarator-named function ranges from a parsed file; anonymous functions keep ``<anon>`` only when nothing names them."""
|
||||
lines = text.splitlines()
|
||||
ranges: list[FunctionRange] = []
|
||||
for fn in ir.functions:
|
||||
if fn.name == MODULE_FUNCTION:
|
||||
continue
|
||||
line = lines[fn.start_line - 1] if 0 < fn.start_line <= len(lines) else ""
|
||||
declared = declarator_name(line)
|
||||
name = declared or fn.name
|
||||
if name in RESERVED_FUNCTION_LABELS and name != ANON_FUNCTION:
|
||||
name = ANON_FUNCTION
|
||||
elif declared is None and name != ANON_FUNCTION and ir.language in _PARAM_NAMED_ARROWS and not _declares(line, name):
|
||||
name = ANON_FUNCTION # tree-sitter named a bare-parameter arrow (`block => {`) after its parameter
|
||||
ranges.append((name, fn.start_line, fn.end_line))
|
||||
return tuple(ranges)
|
||||
|
||||
|
||||
def _declares(line: str, name: str) -> bool:
|
||||
"""True when ``line`` visibly declares ``name`` as a function (keyword form or ``name(``), not as a parameter."""
|
||||
escaped = re.escape(name)
|
||||
return bool(re.search(r"\b(?:function|def|sub)\s+" + escaped + r"\b|(?<![\w$.])" + escaped + r"\s*\(", line))
|
||||
|
||||
|
||||
def named_function_ranges(path: str, text: str, language: str) -> tuple[FunctionRange, ...] | None:
|
||||
"""Ranges for one file. None when the parser rejected the file; regex line scan only when no grammar exists."""
|
||||
ir = parse_file(path, text, language)
|
||||
if ir.parse_ok:
|
||||
return named_ranges_from_ir(ir, text)
|
||||
if ir.reason == "parse_failed":
|
||||
return None
|
||||
lines = text.splitlines()
|
||||
fallback: list[FunctionRange] = []
|
||||
for index, line in enumerate(lines, start=1):
|
||||
name = declarator_name(line)
|
||||
if name:
|
||||
fallback.append((name, index, len(lines)))
|
||||
return tuple(fallback)
|
||||
|
||||
|
||||
def function_at(ranges: Sequence[FunctionRange], line: int | None) -> str | None:
|
||||
"""Innermost *named* function whose range contains ``line``; None when unknown or only anonymous."""
|
||||
if line is None:
|
||||
return None
|
||||
best: FunctionRange | None = None
|
||||
for item in ranges:
|
||||
name, start, end = item
|
||||
if name in UNNAMED_FUNCTIONS or not (start <= line <= end):
|
||||
continue
|
||||
if best is None or (end - start) < (best[2] - best[1]):
|
||||
best = item
|
||||
return best[0] if best is not None else None
|
||||
|
||||
|
||||
def spans_named(ranges: Sequence[FunctionRange], function: str) -> tuple[tuple[int, int], ...]:
|
||||
return tuple((start, end) for name, start, end in ranges if name == function)
|
||||
def function_at(ranges: tuple[FunctionRange, ...], line: int | None) -> str:
|
||||
"""kept one release; use source.enclosing"""
|
||||
return enclosing(ranges, line)
|
||||
|
||||
@@ -11,6 +11,7 @@ import tomllib
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from ...source import canonical_language
|
||||
from ..facts import _ruleset_state
|
||||
|
||||
OPERATION_KINDS = ("protected_read", "protected_write", "privileged_action", "security_setting")
|
||||
@@ -18,7 +19,6 @@ DISCHARGER_KINDS = ("path_guard", "identity_constraint", "ownership_check", "non
|
||||
PROVENANCE_KINDS = ("authenticated_context", "request_input", "constant", "unknown")
|
||||
SENSITIVITIES = ("low", "medium", "high")
|
||||
DEFAULT_OBLIGATION_FACTS_DIR = Path(__file__).resolve().parents[2] / "ruleset" / "obligations"
|
||||
_LANGUAGE_ALIASES = {"c_cpp": "c", "cpp": "c", "python_web": "python", "node": "javascript", "ts": "typescript"}
|
||||
|
||||
_OPERATION_FIELDS = frozenset({"id", "kind", "calls", "resource_arg", "requires", "sensitivity", "query_document_arg"})
|
||||
_DISCHARGER_FIELDS = frozenset(
|
||||
@@ -67,7 +67,9 @@ class ObligationFacts:
|
||||
dischargers: tuple[DischargerFact, ...]
|
||||
|
||||
def for_language(self, language: str) -> ObligationFacts:
|
||||
key = _LANGUAGE_ALIASES.get(language, language)
|
||||
key = canonical_language(language)
|
||||
if key == "cpp":
|
||||
key = "c"
|
||||
return ObligationFacts(
|
||||
version=self.version,
|
||||
operations=tuple(item for item in self.operations if item.language == key),
|
||||
|
||||
@@ -15,8 +15,8 @@ import re
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ...source import ranges_from_ir
|
||||
from ..facts import SemanticFacts
|
||||
from ..functions import named_ranges_from_ir
|
||||
from ..ir import CallSite, FileIR, FunctionIR, trailing_name
|
||||
from .facts import DISCHARGER_KINDS, OPERATION_KINDS, PROVENANCE_KINDS, DischargerFact, ObligationFacts, OperationFact
|
||||
|
||||
@@ -382,7 +382,7 @@ def _function_named(ir: FileIR, function: str, text: str) -> FunctionIR | None:
|
||||
if item.name == function:
|
||||
return item
|
||||
if text:
|
||||
for name, start, _end in named_ranges_from_ir(ir, text):
|
||||
for name, start, _end, _kind in ranges_from_ir(ir, text):
|
||||
if name == function:
|
||||
for item in ir.functions:
|
||||
if item.start_line == start:
|
||||
|
||||
@@ -10,9 +10,10 @@ from pathlib import Path
|
||||
from ..findings import SEVERITY_LABEL, StaticFinding
|
||||
from ..policy import load_policy, resolve_severity
|
||||
from ..preprocess import FileTarget
|
||||
from ..source import FunctionRange, ranges_from_ir
|
||||
from ..source import enclosing as function_at
|
||||
from .engines import joern_available, joern_flow
|
||||
from .facts import FactLoadError, SemanticFacts, load_facts
|
||||
from .functions import FunctionRange, function_at, named_ranges_from_ir
|
||||
from .ir import parse_file
|
||||
from .taint import Demotion, FileFlow, TaintPath, UnadjudicatedFlow, analyze_ir, merge_joern_paths
|
||||
|
||||
@@ -172,7 +173,7 @@ def _flows_for(
|
||||
if not ir.parse_ok:
|
||||
flows[target.path] = UnadjudicatedFlow(path=target.path, reason=ir.reason or "parse_failed", engine=ir.engine)
|
||||
continue
|
||||
ranges[target.path] = named_ranges_from_ir(ir, text)
|
||||
ranges[target.path] = ranges_from_ir(ir, text)
|
||||
analyzed: FileFlow | UnadjudicatedFlow = analyze_ir(ir, facts.for_language(target.language))
|
||||
if isinstance(analyzed, FileFlow):
|
||||
extra = tuple(joern_extra) if joern_extra is not None else _joern_paths(target.path, text, target.language)
|
||||
|
||||
@@ -0,0 +1,739 @@
|
||||
"""Facts about repository source text, shared by scans, learning and plane tasks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
from collections.abc import Iterable, Mapping, Sequence
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, NamedTuple
|
||||
|
||||
from .preprocess import IGNORED_DIRS, build_file_target, detect_language
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .mapping import EntryPointRecord
|
||||
from .semantic.ir import FileIR
|
||||
|
||||
LANGUAGE_ALIASES: Mapping[str, str] = MappingProxyType(
|
||||
{
|
||||
"c_cpp": "cpp",
|
||||
"c++": "cpp",
|
||||
"python_web": "python",
|
||||
"py": "python",
|
||||
"node": "javascript",
|
||||
"js": "javascript",
|
||||
"ts": "typescript",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def module_function(name: str) -> str:
|
||||
"""Map the legacy module-level sentinel in stored records to the one name the code uses."""
|
||||
return MODULE_FUNCTION if name == LEGACY_MODULE_FUNCTION else name
|
||||
|
||||
|
||||
def canonical_language(name: str) -> str:
|
||||
"""Return the language name used by source parsers."""
|
||||
name = name.strip().lower()
|
||||
return LANGUAGE_ALIASES.get(name, name)
|
||||
|
||||
|
||||
MODULE_FUNCTION = "<module>"
|
||||
LEGACY_MODULE_FUNCTION = "<global>" # what frozen records before 2026-10-13 call module-level code
|
||||
ANON_FUNCTION = "<anon>"
|
||||
MAX_FILE_BYTES = 1_000_000
|
||||
LANGUAGES = frozenset({"python", "php", "javascript", "typescript", "java", "go", "ruby"}) # the declared languages
|
||||
_EXCLUDED_DIRS = frozenset(
|
||||
IGNORED_DIRS | {"vendor", "node_modules", "bower_components", "third_party", "third-party", "dist", "build", "out", "target"}
|
||||
)
|
||||
_NOT_PRODUCT = re.compile(
|
||||
r"(^|/)(tests?|__tests__|__mocks__|spec|specs|docs?|examples?|fixtures?|e2e|cypress|storybook|testdata)/"
|
||||
r"|\.(test|spec|stories|mock|e2e)\.[cm]?[jt]sx?$"
|
||||
r"|_test\.(py|go|rb)$|(^|/)test_[^/]*\.(py|rb)$|Test\.java$|Tests\.java$|_spec\.rb$"
|
||||
r"|\.min\.(js|css)$|\.bundle\.js$"
|
||||
)
|
||||
|
||||
_JS_MODIFIERS = r"(?:(?:export|default|public|private|protected|static|async|readonly|override)\s+)*"
|
||||
_JAVA_MODIFIERS = r"(?:(?:public|private|protected|static|final|abstract|synchronized|native|default|strictfp)\s+)*"
|
||||
DECLARATION: dict[str, re.Pattern[str]] = {
|
||||
"python": re.compile(r"^\s*(?:async\s+)?def\s+(\w+)\s*\("),
|
||||
"php": re.compile(r"^\s*(?:(?:public|private|protected|static|final|abstract)\s+)*function\s+&?(\w+)\s*\("),
|
||||
"javascript": re.compile(
|
||||
r"^\s*" + _JS_MODIFIERS + r"function\s*\*?\s*(\w+)\s*\("
|
||||
r"|^\s*" + _JS_MODIFIERS + r"(\w+)\s*\([^;]*\)\s*(?::[^{=]+)?\{\s*$"
|
||||
r"|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:function\b|\([^)]*\)\s*(?::[^=]+)?=>|\w+\s*=>)"
|
||||
),
|
||||
"java": re.compile(
|
||||
r"^\s*" + _JAVA_MODIFIERS + r"(?:<[^>]*>\s+)?(?!return\b|new\b|throw\b)[\w<>\[\],.?]+\s+(\w+)\s*\([^;]*\)\s*"
|
||||
r"(?:throws\s+[\w.,\s]+)?\{?\s*$"
|
||||
),
|
||||
"go": re.compile(r"^\s*func\s+(?:\([^)]*\)\s*)?(\w+)\s*\("),
|
||||
"ruby": re.compile(r"^\s*def\s+(?:self\.)?(\w+[?!=]?)"),
|
||||
}
|
||||
DECLARATION["typescript"] = DECLARATION["javascript"]
|
||||
KEYWORDS = frozenset(
|
||||
{
|
||||
"if",
|
||||
"for",
|
||||
"while",
|
||||
"switch",
|
||||
"catch",
|
||||
"return",
|
||||
"function",
|
||||
"else",
|
||||
"do",
|
||||
"try",
|
||||
"with",
|
||||
"elif",
|
||||
"except",
|
||||
"new",
|
||||
"throw",
|
||||
"super",
|
||||
"constructor",
|
||||
"foreach",
|
||||
"match",
|
||||
"case",
|
||||
"unless",
|
||||
"until",
|
||||
}
|
||||
)
|
||||
_COMMENT_PREFIXES = ("#", "//", "/*", "*", "*/", "--")
|
||||
|
||||
|
||||
RESERVED_FUNCTION_LABELS = KEYWORDS | frozenset(
|
||||
{
|
||||
"function",
|
||||
"def",
|
||||
"return",
|
||||
"if",
|
||||
"else",
|
||||
"for",
|
||||
"while",
|
||||
"switch",
|
||||
"catch",
|
||||
"class",
|
||||
"async",
|
||||
"await",
|
||||
"lambda",
|
||||
"const",
|
||||
"let",
|
||||
"var",
|
||||
"export",
|
||||
"import",
|
||||
"module",
|
||||
"exports",
|
||||
"new",
|
||||
"this",
|
||||
"self",
|
||||
"anonymous",
|
||||
ANON_FUNCTION,
|
||||
}
|
||||
)
|
||||
|
||||
# KEYWORDS is the line engine's statement-word filter (a declaration named `class` or `self` is still a declaration
|
||||
# for the parser engine, which uses RESERVED_FUNCTION_LABELS). The two sets differ on purpose; see the label diff gate.
|
||||
|
||||
_IDENT = r"[A-Za-z_$][\w$]*"
|
||||
_ARROW_TAIL = r"(?:async\s*)?(?:\([^)]*\)|" + _IDENT + r")\s*=>"
|
||||
_DECL = re.compile(
|
||||
# def name( / function name( / export async function name(
|
||||
r"^\s*(?:export\s+)?(?:async\s+)?(?:def|function|sub)\s+(" + _IDENT + r")\s*[\({]|"
|
||||
# const name = [a.b.c =]* [async] (args) => | ident =>
|
||||
r"^\s*(?:export\s+)?(?:const|let|var)\s+(" + _IDENT + r")\s*=\s*(?:" + _IDENT + r"(?:\." + _IDENT + r")*\s*=\s*)*" + _ARROW_TAIL + r"|"
|
||||
# exports.name = / module.exports.name = [async] function | (args) =>
|
||||
r"^\s*(?:module\.)?exports\.(" + _IDENT + r")\s*=\s*(?:async\s*)?(?:function\b|" + _ARROW_TAIL + r")|"
|
||||
# object property: name: [async] function | (args) =>
|
||||
r"^\s*(?:async\s+)?(" + _IDENT + r")\s*:\s*(?:async\s*)?(?:function\b|" + _ARROW_TAIL + r")|"
|
||||
# C / Java / JS method or shorthand: [type] name(args) { (no call: the arg list carries no callback or literal)
|
||||
r"^\s*(?:[A-Za-z_][\w:<>\*\s,]*[\s\*])?(" + r"[A-Za-z_][\w:$]*" + r")\s*\(([^;{]*)\)\s*(?:const\s*)?\{?\s*$"
|
||||
)
|
||||
_CALLBACK_ARGS = re.compile(r"\bfunction\b|=>|['\"`]|^\s*\d")
|
||||
# A statement keyword before `name(` means the line calls or controls, it does not declare.
|
||||
_STATEMENT_PREFIX = re.compile(
|
||||
r"^\s*(?:return|print|yield|throw|await|new|else|case|delete|typeof|synchronized|with|elif|except|try|if|for|while|switch|catch|do)\b"
|
||||
)
|
||||
# Only these grammars name an arrow function after its bare parameter; elsewhere the parser name is trustworthy.
|
||||
_PARAM_NAMED_ARROWS = frozenset({"javascript", "typescript"})
|
||||
|
||||
|
||||
def declarator_name(line: str, language: str = "") -> str | None:
|
||||
"""Function name declared on ``line``, or None when the line declares nothing (a call is not a declaration)."""
|
||||
pattern = DECLARATION.get(canonical_language(language))
|
||||
if pattern is not None and (match := pattern.match(line)):
|
||||
name = next((group for group in match.groups() if group), "")
|
||||
if name and name not in RESERVED_FUNCTION_LABELS:
|
||||
return name
|
||||
if language in {"python", "ruby", "go", "php", "java"}:
|
||||
return None
|
||||
match = _DECL.match(line)
|
||||
if not match:
|
||||
return None
|
||||
groups = match.groups()
|
||||
name = next((group for group in groups[:5] if group), None)
|
||||
if name is None:
|
||||
return None
|
||||
if groups[4] is not None:
|
||||
args = groups[5] or ""
|
||||
if _CALLBACK_ARGS.search(args):
|
||||
return None # `exec('...', function (err) {` is a call, not a declaration
|
||||
if _STATEMENT_PREFIX.match(line) or name in {"if", "for", "while", "switch", "catch", "return", "elif", "with", "synchronized"}:
|
||||
return None
|
||||
name = name.rsplit("::", 1)[-1] # `bool K::m(` declares `m`; catalog labels are unqualified
|
||||
return name if name not in RESERVED_FUNCTION_LABELS else None
|
||||
|
||||
|
||||
class FunctionRange(NamedTuple):
|
||||
"""A function's name and inclusive, one-based source lines."""
|
||||
|
||||
name: str
|
||||
start: int
|
||||
end: int
|
||||
kind: str
|
||||
|
||||
|
||||
class CallSite(NamedTuple):
|
||||
"""A call's one-based line and the name it invokes."""
|
||||
|
||||
line: int
|
||||
name: str
|
||||
|
||||
|
||||
class SourceFacts(NamedTuple):
|
||||
"""Repository evidence about one source location."""
|
||||
|
||||
language: str
|
||||
function: str
|
||||
span: tuple[int, int] | None
|
||||
callers: tuple[tuple[str, int], ...]
|
||||
entry_points: tuple[EntryPointRecord, ...]
|
||||
|
||||
|
||||
def product_files(root: Path) -> list[str]:
|
||||
"""Return sorted repository-relative product paths, excluding tests and dependencies."""
|
||||
from .model.layout import is_test_path, is_vendored, layout_facts
|
||||
|
||||
root = Path(root)
|
||||
facts = layout_facts()
|
||||
files = []
|
||||
for path in sorted(root.rglob("*")):
|
||||
relative = path.relative_to(root)
|
||||
if any(part.startswith(".") or part in _EXCLUDED_DIRS for part in relative.parts):
|
||||
continue
|
||||
if path.is_symlink() or any(parent.is_symlink() for parent in path.parents if parent != root) or not path.is_file():
|
||||
continue
|
||||
posix = relative.as_posix()
|
||||
if _NOT_PRODUCT.search(posix) or is_vendored(posix, facts) or is_test_path(posix, facts):
|
||||
continue
|
||||
if detect_language(path) in LANGUAGES and path.stat().st_size <= MAX_FILE_BYTES:
|
||||
files.append(posix)
|
||||
return files
|
||||
|
||||
|
||||
def ranges_from_ir(ir: FileIR, text: str) -> tuple[FunctionRange, ...]:
|
||||
"""Use declarator names when a parser names a function after its parameter."""
|
||||
lines = text.splitlines()
|
||||
result = []
|
||||
for fn in ir.functions:
|
||||
if fn.name == MODULE_FUNCTION:
|
||||
continue
|
||||
line = lines[fn.start_line - 1] if 0 < fn.start_line <= len(lines) else ""
|
||||
declared = declarator_name(line, ir.language)
|
||||
name = declared or fn.name
|
||||
if (
|
||||
name in RESERVED_FUNCTION_LABELS
|
||||
and name != ANON_FUNCTION
|
||||
or declared is None
|
||||
and ir.language in _PARAM_NAMED_ARROWS
|
||||
and not _declares(line, name)
|
||||
):
|
||||
name = ANON_FUNCTION
|
||||
result.append(FunctionRange(name, fn.start_line, fn.end_line, "anonymous" if name == ANON_FUNCTION else "function"))
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def _declares(line: str, name: str) -> bool:
|
||||
escaped = re.escape(name)
|
||||
return bool(re.search(r"\b(?:function|def|sub)\s+" + escaped + r"\b|(?<![\w$.])" + escaped + r"\s*\(", line))
|
||||
|
||||
|
||||
def _code_lines(text: str) -> list[str]:
|
||||
"""Blank comments and quoted strings before counting delimiters."""
|
||||
token = re.compile(r"/\*[\s\S]*?\*/|//[^\n]*|\#[^\n]*|'(?:\\.|[^'\\])*'|\"(?:\\.|[^\"\\])*\"|`(?:\\.|[^`\\])*`")
|
||||
return token.sub(lambda match: re.sub(r"[^\n]", " ", match.group()), text).splitlines()
|
||||
|
||||
|
||||
def _body_end(code: Sequence[str], start: int, column: int) -> int | None:
|
||||
"""Find the end of a body after its declaration head, skipping prototypes and calls."""
|
||||
parens, depth, opened, arrow = 0, 0, False, False
|
||||
for row in range(start, len(code)):
|
||||
for col in range(column if row == start else 0, len(code[row])):
|
||||
char = code[row][col]
|
||||
if opened:
|
||||
if char == "{":
|
||||
depth += 1
|
||||
elif char == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return row + 1
|
||||
elif char == "(":
|
||||
parens += 1
|
||||
elif char == ")":
|
||||
parens -= 1
|
||||
if parens < 0:
|
||||
return None
|
||||
elif parens == 0:
|
||||
if code[row].startswith("=>", col):
|
||||
arrow = True
|
||||
if char == "{":
|
||||
opened, depth = True, 1
|
||||
elif char in ";}":
|
||||
return row + 1 if arrow else None
|
||||
elif char == "," and arrow:
|
||||
return row + 1
|
||||
if not opened and arrow:
|
||||
return row + 1
|
||||
if not opened and row - start >= 40:
|
||||
return None
|
||||
return len(code) if opened else None
|
||||
|
||||
|
||||
_HEAD = re.compile(
|
||||
r"(?<![\w.$>])([A-Za-z_$][\w:$]*)\s*(?:<[^<>(){};]*>\s*)?(?=\()|^\s*sub\s+(\w+)\s*(?=\{)"
|
||||
r"|(?<![\w$])([A-Za-z_$][\w$]*)\s*[:=]\s*(?:[\w$.]+\s*=\s*)*(?:async\s*)?(?:function\b\s*[\w$]*\s*)?(?=\(|[\w$]+\s*=>)"
|
||||
)
|
||||
|
||||
|
||||
def _regex_ranges(text: str, language: str) -> tuple[FunctionRange, ...]:
|
||||
lines, code = text.splitlines(), _code_lines(text)
|
||||
result = []
|
||||
for index, line in enumerate(code):
|
||||
name = declarator_name(line, language)
|
||||
if language == "python":
|
||||
if not name:
|
||||
continue
|
||||
indent = len(line) - len(line.lstrip())
|
||||
end = index + 1
|
||||
for row in range(index + 1, len(code)):
|
||||
if code[row].strip() and len(code[row]) - len(code[row].lstrip()) <= indent:
|
||||
break
|
||||
if code[row].strip():
|
||||
end = row + 1
|
||||
elif language == "ruby":
|
||||
if not name:
|
||||
continue
|
||||
depth, end = 1, len(lines)
|
||||
for row in range(index + 1, len(code)):
|
||||
if re.match(r"\s*(?:def|class|module|if|unless|case|begin|while|until|for)\b", code[row]):
|
||||
depth += 1
|
||||
if re.search(r"\bdo\s*(?:\|[^|]*\|)?\s*$", code[row]):
|
||||
depth += 1
|
||||
if re.match(r"\s*end\b", code[row]):
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
end = row + 1
|
||||
break
|
||||
else:
|
||||
if _STATEMENT_PREFIX.match(line):
|
||||
continue
|
||||
heads = list(_HEAD.finditer(line))
|
||||
if not heads:
|
||||
continue
|
||||
head = heads[0]
|
||||
candidate = next(group for group in head.groups() if group).rsplit("::", 1)[-1]
|
||||
name = name or candidate
|
||||
if name in RESERVED_FUNCTION_LABELS:
|
||||
continue
|
||||
end = _body_end(code, index, head.end())
|
||||
if end is None:
|
||||
continue
|
||||
result.append(FunctionRange(name, index + 1, end, "function"))
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def ranges(path: str, text: str, language: str) -> tuple[FunctionRange, ...]:
|
||||
"""Return inclusive source ranges using tree-sitter, or a declaration scan without a grammar.
|
||||
|
||||
A parser rejection raises ValueError so unreadable syntax cannot look like a file without functions; a language
|
||||
without a grammar uses the declaration scan.
|
||||
"""
|
||||
from .semantic.cst import parse_with_cst
|
||||
|
||||
language = canonical_language(language)
|
||||
ir = parse_with_cst(path, text, language)
|
||||
if ir is not None:
|
||||
if not ir.parse_ok and ir.reason != "language_unsupported":
|
||||
raise ValueError(f"{path}: source parse failed")
|
||||
named = ranges_from_ir(ir, text) if ir.parse_ok else _regex_ranges(text, language)
|
||||
else:
|
||||
named = _regex_ranges(text, language)
|
||||
module = (FunctionRange(MODULE_FUNCTION, 1, len(text.splitlines()), "module"),) if text.splitlines() else ()
|
||||
return module + named
|
||||
|
||||
|
||||
def enclosing(ranges: Sequence[FunctionRange], line: int | None) -> str:
|
||||
"""Return the innermost named function at a one-based line, or the module label."""
|
||||
found = [item for item in ranges if item.kind == "function" and line is not None and item.start <= line <= item.end]
|
||||
return min(found, key=lambda item: item.end - item.start).name if found else MODULE_FUNCTION
|
||||
|
||||
|
||||
def bounds(ranges: Sequence[FunctionRange], name: str) -> tuple[int, int] | None:
|
||||
"""Return the first matching function's inclusive, one-based bounds."""
|
||||
name = re.split(r"::|\.|->", name)[-1]
|
||||
return next(((item.start, item.end) for item in ranges if item.name == name), None)
|
||||
|
||||
|
||||
def spans_named(ranges: Sequence[FunctionRange], name: str) -> tuple[tuple[int, int], ...]:
|
||||
"""Return all ranges of a name, including repeated method declarations."""
|
||||
return tuple((item.start, item.end) for item in ranges if item.name == name)
|
||||
|
||||
|
||||
def call_sites(text: str, names: Iterable[str]) -> tuple[CallSite, ...]:
|
||||
"""Return calls in line order, excluding declarations and comments, once per name and line."""
|
||||
wanted = sorted(set(names) - RESERVED_FUNCTION_LABELS - {MODULE_FUNCTION}, key=lambda name: (-len(name), name))
|
||||
if not wanted:
|
||||
return ()
|
||||
token = re.compile(r"(?<![A-Za-z0-9_$])(" + "|".join(re.escape(name) for name in wanted) + r")\s*\(")
|
||||
result = []
|
||||
for number, line in enumerate(_code_lines(text), start=1):
|
||||
declared = {
|
||||
next((group for group in match.groups() if group), "") for pattern in DECLARATION.values() if (match := pattern.match(line))
|
||||
}
|
||||
if "{" in line or "=>" in line:
|
||||
declared.add(declarator_name(line))
|
||||
seen = set()
|
||||
for match in token.finditer(line):
|
||||
name = match.group(1)
|
||||
# A same-line body can call its own function after the declaration head.
|
||||
if name in seen or (name in declared and not any(char in line[: match.start()] for char in "{;")):
|
||||
continue
|
||||
seen.add(name)
|
||||
result.append(CallSite(number, name))
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def _read_source(root: Path, path: str) -> str:
|
||||
resolved = (root / path).resolve()
|
||||
if not resolved.is_relative_to(root.resolve()) or any(part.startswith(".env") for part in Path(path).parts):
|
||||
raise ValueError(f"source path is outside the repository or private: {path}")
|
||||
return resolved.read_text(encoding="utf-8", errors="replace")
|
||||
|
||||
|
||||
def callers_of(root: Path, name: str, *, files: Iterable[str]) -> tuple[tuple[str, int], ...]:
|
||||
"""Return sorted call locations from the supplied repository-relative product files."""
|
||||
root = Path(root)
|
||||
return tuple((path, call.line) for path in sorted(set(files)) for call in call_sites(_read_source(root, path), (name,)))
|
||||
|
||||
|
||||
def facts_for(root: Path, path: str, line: int, *, entry_points: Sequence[EntryPointRecord] | None = None) -> SourceFacts:
|
||||
"""Compute repository facts at a line; supplied entry points avoid repeated mapping."""
|
||||
root = Path(root)
|
||||
text = _read_source(root, path)
|
||||
language = canonical_language(detect_language(root / path))
|
||||
spans = ranges(path, text, language)
|
||||
function = enclosing(spans, line)
|
||||
files = product_files(root)
|
||||
if entry_points is None:
|
||||
from .mapping import analyze_entry_points
|
||||
|
||||
entry_points = analyze_entry_points(root, [build_file_target(root, root / file) for file in files])
|
||||
return SourceFacts(language, function, bounds(spans, function), callers_of(root, function, files=files), tuple(entry_points))
|
||||
|
||||
|
||||
def _digest(names: Sequence[str]) -> str:
|
||||
"""Identify a sorted set of function names for stored facts."""
|
||||
return hashlib.sha256("\n".join(names).encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
# --- the line-pattern engine: declarations by regular expression, spans to the next declaration, brace matching -----
|
||||
# This is what labels, features, roles, excerpts and the plane facts were built on. The parser-based `ranges` above
|
||||
# serves the scorer and the overlay; both share DECLARATION, KEYWORDS, RESERVED_FUNCTION_LABELS and the sentinel.
|
||||
|
||||
HASH_COMMENTS = frozenset({"python", "php", "ruby", "shell", "perl", "yaml", "r"})
|
||||
|
||||
SLASH_COMMENTS = frozenset(
|
||||
{"c", "cpp", "java", "javascript", "typescript", "php", "go", "groovy", "csharp", "kotlin", "rust", "swift", "scala"}
|
||||
) # fmt: skip
|
||||
|
||||
def _comment_ranges(line: str, state: str | None, language: str) -> tuple[list[tuple[int, int]], str | None]:
|
||||
"""The comment spans of one line and the block state carried to the next (``block`` inside ``/* */``, a triple
|
||||
quote inside a Python docstring). Strings are skipped so a ``#`` in a literal is not a comment."""
|
||||
hashes, slashes, python = language in HASH_COMMENTS, language in SLASH_COMMENTS, language == "python"
|
||||
ranges: list[tuple[int, int]] = []
|
||||
i, n, quote = 0, len(line), None
|
||||
if state is not None:
|
||||
close = "*/" if state == "block" else state
|
||||
j = line.find(close)
|
||||
if j < 0:
|
||||
return [(0, n)], state
|
||||
ranges.append((0, j + len(close)))
|
||||
i, state = j + len(close), None
|
||||
while i < n:
|
||||
c = line[i]
|
||||
if quote:
|
||||
if c == "\\":
|
||||
i += 2
|
||||
continue
|
||||
if c == quote:
|
||||
quote = None
|
||||
i += 1
|
||||
continue
|
||||
if python and line.startswith(('"""', "'''"), i) and not line[:i].strip():
|
||||
delim = line[i : i + 3]
|
||||
j = line.find(delim, i + 3)
|
||||
if j < 0:
|
||||
ranges.append((i, n))
|
||||
return ranges, delim
|
||||
ranges.append((i, j + 3))
|
||||
i = j + 3
|
||||
continue
|
||||
if c in "\"'`":
|
||||
quote = c
|
||||
elif (slashes and line.startswith("//", i)) or (hashes and c == "#"):
|
||||
ranges.append((i, n))
|
||||
break
|
||||
elif slashes and line.startswith("/*", i):
|
||||
j = line.find("*/", i + 2)
|
||||
if j < 0:
|
||||
ranges.append((i, n))
|
||||
return ranges, "block"
|
||||
ranges.append((i, j + 2))
|
||||
i = j + 2
|
||||
continue
|
||||
i += 1
|
||||
return ranges, state
|
||||
|
||||
_UNKNOWN_LANGUAGES = frozenset({"", "other", "unknown"})
|
||||
|
||||
_PERL_LINE = re.compile(r"^\s*(?:sub\s+\w+\s*[{(;]|use\s+strict\s*;|my\s+[$@%]\w+)")
|
||||
|
||||
_RUBY_DEF = re.compile(r"^\s*def\s+(?:self\.)?\w+[?!=]?(?:\s*\(|\s|$)")
|
||||
|
||||
_RUBY_END = re.compile(r"^\s*end\s*$")
|
||||
|
||||
_BODY_SCAN_LINES = 40 # how far past a declaration head the body's ``{`` (or a prototype's ``;``) may be
|
||||
|
||||
_STATEMENT_HEAD = re.compile(r"\s*(return|if|while|for|switch|else|case|throw|await|yield)\b")
|
||||
|
||||
_FUNCTION_EXPR = r"(?:async\s+)?(?:function\b\s*\*?\s*[\w$]*\s*)?(?=\(|[\w$]+\s*=>)" # function (...) / (...) => / a =>
|
||||
|
||||
def resolve_language(lines: Sequence[str], language: str) -> str:
|
||||
"""The alias-normalised language; a file given as unknown (``other``, the feature vocabulary's bucket for Go, Ruby,
|
||||
C#, Perl, ...) is sniffed for Perl (``sub name {``, ``use strict;``) and Ruby (``def`` ... ``end``), whose
|
||||
declaration forms a brace matcher never finds and whose ``#`` comments must be redacted."""
|
||||
language = canonical_language(language)
|
||||
if language in _UNKNOWN_LANGUAGES and lines:
|
||||
if any(_PERL_LINE.match(line) for line in lines):
|
||||
return "perl"
|
||||
if any(_RUBY_DEF.match(line) for line in lines) and any(_RUBY_END.match(line) for line in lines):
|
||||
return "ruby"
|
||||
return language
|
||||
|
||||
def _blank_comments(lines: Sequence[str], language: str) -> list[str]:
|
||||
style = language if language in SLASH_COMMENTS or language in HASH_COMMENTS else "c"
|
||||
code: list[str] = []
|
||||
state: str | None = None
|
||||
for line in lines:
|
||||
ranges, state = _comment_ranges(line, state, style)
|
||||
for start, end in reversed(ranges):
|
||||
line = line[:start] + " " * (end - start) + line[end:]
|
||||
code.append(line)
|
||||
return code
|
||||
|
||||
def _declaration_heads(name: str) -> re.Pattern[str]:
|
||||
"""Where a brace language may declare ``name`` (the body follows): ``name(`` (a definition -- or a call, told
|
||||
apart by :func:`_body_start`); Perl ``sub name {``; an object-literal method ``name: function (`` /
|
||||
``name: (a) =>``; an assignment ``x.y.name = function (`` / ``= (a) =>`` / ``const name = m.name = async a =>``
|
||||
(any chain). ``$`` is an identifier character (``Glance$serveRequest``)."""
|
||||
quoted = re.escape(name)
|
||||
return re.compile(
|
||||
rf"(?<![\w.>$]){quoted}\s*(?:<[^<>(){{}};]*>\s*)?(?=\()"
|
||||
rf"|^\s*sub\s+{quoted}\s*(?=[{{(;])"
|
||||
rf"|(?<![\w$]){quoted}\s*:\s*{_FUNCTION_EXPR}"
|
||||
rf"|(?<![\w$]){quoted}\s*=\s*(?:[\w$.\[\]'\"]+\s*=\s*)*{_FUNCTION_EXPR}"
|
||||
)
|
||||
|
||||
def _body_start(code: Sequence[str], line: int, column: int) -> tuple[str, int, int] | None:
|
||||
"""Scan from a declaration head for its body: ``("brace", line, column)`` of the ``{`` at parenthesis depth 0,
|
||||
``("expression", line, column)`` of the ``;`` or ``,`` ending an arrow function's expression body, ``None`` for a
|
||||
prototype (``;`` before any body), a call (a ``)`` closing an outer parenthesis, a ``}``) or no body within
|
||||
:data:`_BODY_SCAN_LINES`. Parentheses are tracked so ``f(a, function () {})`` is not a definition of ``f``."""
|
||||
depth, arrow = 0, False
|
||||
for j in range(line, min(line + _BODY_SCAN_LINES, len(code))):
|
||||
text = code[j]
|
||||
k = column if j == line else 0
|
||||
while k < len(text):
|
||||
c = text[k]
|
||||
if c == "(":
|
||||
depth += 1
|
||||
elif c == ")":
|
||||
depth -= 1
|
||||
if depth < 0:
|
||||
return None
|
||||
elif depth == 0:
|
||||
if c == "{":
|
||||
return ("brace", j, k)
|
||||
if arrow and c in ";,":
|
||||
return ("expression", j, k)
|
||||
if c in "};": # a statement ends (a call, a prototype) before any body opened
|
||||
return None
|
||||
if text.startswith("=>", k):
|
||||
arrow = True
|
||||
k += 1
|
||||
k += 1
|
||||
return None
|
||||
|
||||
def _brace_bounds(lines: Sequence[str], language: str, function: str) -> tuple[int, int] | None:
|
||||
"""A brace language's definition of ``function`` (``Class::name`` and ``a.b`` reduced to the last name): the first
|
||||
declaration head (:func:`_declaration_heads`) whose body :func:`_body_start` finds, to the matching ``}`` (or the
|
||||
end of an expression body). Comments are blanked before anything is matched or counted."""
|
||||
name = re.split(r"::|\.|->", function)[-1]
|
||||
if not re.fullmatch(r"[A-Za-z_$][\w$]*", name):
|
||||
return None
|
||||
heads = _declaration_heads(name)
|
||||
code = _blank_comments(lines, language)
|
||||
for i, line in enumerate(code):
|
||||
if _STATEMENT_HEAD.match(line):
|
||||
continue
|
||||
for found in heads.finditer(line):
|
||||
body = _body_start(code, i, found.end())
|
||||
if body is None:
|
||||
continue
|
||||
kind, j, k = body
|
||||
if kind == "expression":
|
||||
return i, j + 1
|
||||
depth = 0
|
||||
for row in range(j, len(code)):
|
||||
for c in code[row][k:] if row == j else code[row]:
|
||||
if c == "{":
|
||||
depth += 1
|
||||
elif c == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return i, row + 1
|
||||
return i, len(code)
|
||||
return None
|
||||
|
||||
def function_bounds(lines: Sequence[str], language: str, function: str) -> tuple[int, int] | None:
|
||||
"""0-based ``[start, end)`` of the first declaration of ``function`` (the whole file for ``<global>``), or
|
||||
``None`` when the file does not declare it. Languages without a repo-facts declaration pattern (C, C++, C#, Perl,
|
||||
...) and brace languages whose pattern misses the name use the brace matcher; an unknown language is sniffed
|
||||
(:func:`resolve_language`)."""
|
||||
function = module_function(function)
|
||||
language = resolve_language(lines, language)
|
||||
if function == MODULE_FUNCTION:
|
||||
return (0, len(lines)) if lines else None
|
||||
pattern = DECLARATION.get(language)
|
||||
start = next((i for i, line in enumerate(lines) if _declared_name(pattern, line) == function), None) if pattern else None
|
||||
if start is None:
|
||||
return _brace_bounds(lines, language, function) if language in SLASH_COMMENTS or pattern is None else None
|
||||
span = function_span(lines, language, function)
|
||||
if not span:
|
||||
return None
|
||||
return start, min(start + span, len(lines))
|
||||
|
||||
def product_paths(workspace: Path) -> list[Path]:
|
||||
"""Product source files of the supported languages, sorted; tests, vendored trees, build output, minified
|
||||
bundles and dot-directories are not product (the classification of `model_sinks.py` and the layout facts)."""
|
||||
from .model.layout import is_test_path, is_vendored, layout_facts
|
||||
|
||||
facts = layout_facts()
|
||||
files: list[Path] = []
|
||||
for path in sorted(workspace.rglob("*")):
|
||||
if path.is_symlink() or not path.is_file():
|
||||
continue
|
||||
relative = path.relative_to(workspace)
|
||||
parts = relative.parts
|
||||
if any(part.startswith(".") or part in _EXCLUDED_DIRS for part in parts[:-1]) or parts[-1].startswith("."):
|
||||
continue
|
||||
posix = relative.as_posix()
|
||||
if _NOT_PRODUCT.search(posix) or is_vendored(posix, facts) or is_test_path(posix, facts):
|
||||
continue
|
||||
if detect_language(path) not in LANGUAGES or path.stat().st_size > MAX_FILE_BYTES:
|
||||
continue
|
||||
files.append(path)
|
||||
return files
|
||||
|
||||
def declared_functions(lines: Sequence[str], language: str) -> list[str]:
|
||||
"""The function names a file declares, in source order, each once."""
|
||||
pattern = DECLARATION.get(language)
|
||||
if pattern is None:
|
||||
return []
|
||||
names: list[str] = []
|
||||
for line in lines:
|
||||
name = _declared_name(pattern, line)
|
||||
if name and name not in names:
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
def enclosing_at(lines: Sequence[str], index: int, language: str) -> str:
|
||||
"""The nearest declaration at or above line ``index`` (0-based), `<global>` when there is none."""
|
||||
pattern = DECLARATION.get(language)
|
||||
if pattern is None:
|
||||
return MODULE_FUNCTION
|
||||
indent = _indent(lines[index]) if language == "python" else None
|
||||
for back in range(index, -1, -1):
|
||||
name = _declared_name(pattern, lines[back])
|
||||
if not name:
|
||||
continue
|
||||
if indent is None or back == index or _indent(lines[back]) < indent:
|
||||
return name
|
||||
indent = min(indent, _indent(lines[back])) # a sibling or deeper def: keep looking for the one that encloses
|
||||
return MODULE_FUNCTION
|
||||
|
||||
def _indent(line: str) -> int:
|
||||
return len(line) - len(line.lstrip(" \t"))
|
||||
|
||||
def call_sites_in(lines: Sequence[str], language: str, names: Iterable[str]) -> list[dict[str, Any]]:
|
||||
"""Calls to any of ``names`` in one file: `{"function", "line", "enclosing"}`, one per name and line, in
|
||||
line order. A declaration line is not a call of the function it declares; comment lines are skipped."""
|
||||
wanted = sorted(set(names), key=lambda n: (-len(n), n))
|
||||
if not wanted:
|
||||
return []
|
||||
token = re.compile(r"(?<![A-Za-z0-9_$])(" + "|".join(re.escape(n) for n in wanted) + r")\s*\(")
|
||||
pattern = DECLARATION.get(language)
|
||||
found: list[dict[str, Any]] = []
|
||||
for index, line in enumerate(lines):
|
||||
if line.lstrip().startswith(_COMMENT_PREFIXES):
|
||||
continue
|
||||
declared = _declared_name(pattern, line) if pattern is not None else ""
|
||||
seen: set[str] = set()
|
||||
for match in token.finditer(line):
|
||||
name = match.group(1)
|
||||
if name == declared or name in seen:
|
||||
continue
|
||||
seen.add(name)
|
||||
found.append({"function": name, "line": index + 1, "enclosing": enclosing_at(lines, index, language)})
|
||||
return found
|
||||
|
||||
def _declared_name(pattern: re.Pattern[str], line: str) -> str:
|
||||
match = pattern.match(line)
|
||||
if not match:
|
||||
return ""
|
||||
name = next((g for g in match.groups() if g), "")
|
||||
return "" if name in KEYWORDS else name
|
||||
|
||||
def function_span(lines: Sequence[str], language: str, function: str) -> int | None:
|
||||
"""Lines of the first declaration of ``function`` up to the next declaration (Python: up to the next line indented
|
||||
no deeper), ``None`` when the file does not declare it. ``<global>`` counts the whole file."""
|
||||
function = module_function(function)
|
||||
if function == MODULE_FUNCTION:
|
||||
return len(lines)
|
||||
pattern = DECLARATION.get(language)
|
||||
if pattern is None:
|
||||
return None
|
||||
start = next((i for i, line in enumerate(lines) if _declared_name(pattern, line) == function), None)
|
||||
if start is None:
|
||||
return None
|
||||
indent = len(lines[start]) - len(lines[start].lstrip(" \t"))
|
||||
for end in range(start + 1, len(lines)):
|
||||
line = lines[end]
|
||||
if language == "python":
|
||||
if line.strip() and not line.lstrip().startswith("#") and len(line) - len(line.lstrip(" \t")) <= indent:
|
||||
return end - start
|
||||
elif _declared_name(pattern, line):
|
||||
return end - start
|
||||
return len(lines) - start
|
||||
|
||||
|
||||
declared_name = _declared_name
|
||||
+2
-2
@@ -12,14 +12,14 @@
|
||||
"degradations": [],
|
||||
"findings": [
|
||||
{
|
||||
"site": "app/index.php:3:<global>",
|
||||
"site": "app/index.php:3:<module>",
|
||||
"family": "injection",
|
||||
"rung": "model_entailed",
|
||||
"witness": "$_GET[\"id\"] -> mysqli_query($conn, \"SELECT * FROM users WHERE id = \" . $id) in app/index.php (line 3, 4 steps)",
|
||||
"reached_from": 1
|
||||
},
|
||||
{
|
||||
"site": "app/index.php:5:<global>",
|
||||
"site": "app/index.php:5:<module>",
|
||||
"family": "injection",
|
||||
"rung": "model_entailed",
|
||||
"witness": "$_GET[\"host\"] -> system(\"ping \" . $_GET[\"host\"]) in app/index.php (line 5, 2 steps)",
|
||||
|
||||
@@ -86,9 +86,10 @@ def test_read_file_truncates_and_rejects_symlink_escape(tmp_path: Path) -> None:
|
||||
|
||||
def test_find_refs_searches_source_text(tmp_path: Path) -> None:
|
||||
root = _repo_with_split_sink(tmp_path)
|
||||
refs = find_refs(root, "query")
|
||||
assert [ref["line"] for ref in refs] == [2, 3]
|
||||
assert all(ref["path"] == "app.py" and ref["name"] == "query" and ref["source"] == "text" for ref in refs)
|
||||
refs = find_refs(root, "execute")
|
||||
assert [ref["line"] for ref in refs] == [3]
|
||||
assert all(ref["path"] == "app.py" and ref["name"] == "execute" and ref["source"] == "text" for ref in refs)
|
||||
assert find_refs(root, "query") == []
|
||||
assert find_refs(root, "search") == []
|
||||
assert find_refs(root, "") == []
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""freeze.py resolves an advisory's file-only site to the function the fix changes (population v3, decision 2).
|
||||
|
||||
The resolver reads PHP text only; it must not be fooled by braces in strings, comments or inline HTML, and a line
|
||||
outside every function is `<global>`."""
|
||||
outside every function is `<module>`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -48,12 +48,12 @@ def test_a_line_resolves_to_its_innermost_named_function_or_global() -> None:
|
||||
spans = freeze.php_functions(PHP)
|
||||
assert freeze.enclosing(spans, 9) == "open" # inside the closure: the named function owns it
|
||||
assert freeze.enclosing(spans, 17) == "close"
|
||||
assert freeze.enclosing(spans, 21) == "<global>"
|
||||
assert freeze.enclosing(spans, 1) == "<global>"
|
||||
assert freeze.enclosing(spans, 21) == "<module>"
|
||||
assert freeze.enclosing(spans, 1) == "<module>"
|
||||
|
||||
|
||||
def test_an_insertion_after_a_closing_brace_is_outside_the_function() -> None:
|
||||
spans = freeze.php_functions(PHP)
|
||||
assert freeze.enclosing(spans, 10) == "open" # a changed line on the closing brace
|
||||
assert freeze.enclosing(spans, 10, inserted=True) == "<global>" # new code that follows it
|
||||
assert freeze.enclosing(spans, 10, inserted=True) == "<module>" # new code that follows it
|
||||
assert freeze.enclosing(spans, 8, inserted=True) == "open"
|
||||
|
||||
+21
-12
@@ -512,12 +512,17 @@ def test_known_limit_pairs_emit_no_improve_signals(tmp_path: Path) -> None:
|
||||
|
||||
|
||||
def test_function_at_skips_anonymous_and_module_ranges() -> None:
|
||||
from openultrasast.semantic.overlay import function_at
|
||||
from openultrasast.source import FunctionRange
|
||||
from openultrasast.source import enclosing as function_at
|
||||
|
||||
ranges = (("<module>", 1, 40), ("handler", 3, 20), ("<anon>", 5, 12))
|
||||
ranges = (
|
||||
FunctionRange("<module>", 1, 40, "module"),
|
||||
FunctionRange("handler", 3, 20, "function"),
|
||||
FunctionRange("<anon>", 5, 12, "anonymous"),
|
||||
)
|
||||
assert function_at(ranges, 7) == "handler" # innermost *named* function, never "<anon>"
|
||||
assert function_at(ranges, 25) is None
|
||||
assert function_at((("<anon>", 1, 9),), 4) is None
|
||||
assert function_at(ranges, 25) == "<module>"
|
||||
assert function_at((FunctionRange("<anon>", 1, 9, "anonymous"),), 4) == "<module>"
|
||||
|
||||
|
||||
@pytest.mark.semantic
|
||||
@@ -738,7 +743,8 @@ def test_unresolvable_function_label_is_a_visible_loss_not_a_silent_miss(tmp_pat
|
||||
|
||||
def test_named_function_ranges_use_declarator_names() -> None:
|
||||
from openultrasast.semantic.extra import has_semantic_extra
|
||||
from openultrasast.semantic.functions import declarator_name, named_function_ranges
|
||||
from openultrasast.source import declarator_name
|
||||
from openultrasast.source import ranges as named_function_ranges
|
||||
|
||||
assert declarator_name("const listProcessesOnPort = module.exports.listProcessesOnPort = async port => {") == "listProcessesOnPort"
|
||||
assert declarator_name(" exec('df -k ' + path, function(err, stdout) {") is None
|
||||
@@ -765,8 +771,8 @@ def test_named_function_ranges_use_declarator_names() -> None:
|
||||
)
|
||||
ranges = named_function_ranges("f.js", src, "javascript")
|
||||
assert ranges is not None
|
||||
assert {name for name, _, _ in ranges} >= {"a", "b", "c", "d", "e"}
|
||||
assert "<anon>" not in {name for name, _, _ in ranges}
|
||||
assert {name for name, _, _, _ in ranges} >= {"a", "b", "c", "d", "e"}
|
||||
assert "<anon>" not in {name for name, _, _, _ in ranges}
|
||||
|
||||
|
||||
# --- review round 4 remediation (RED first) -----------------------------------
|
||||
@@ -775,17 +781,18 @@ def test_named_function_ranges_use_declarator_names() -> None:
|
||||
@pytest.mark.semantic
|
||||
def test_c_function_with_return_type_on_previous_line_keeps_its_parser_name() -> None:
|
||||
from openultrasast.semantic.extra import has_semantic_extra
|
||||
from openultrasast.semantic.functions import function_at, named_function_ranges
|
||||
from openultrasast.source import enclosing as function_at
|
||||
from openultrasast.source import ranges as named_function_ranges
|
||||
|
||||
if not has_semantic_extra():
|
||||
pytest.skip("tree-sitter extra absent")
|
||||
ranges = named_function_ranges("a.c", "static int\nfoo(int x)\n{\n return x;\n}\n", "c")
|
||||
assert ranges is not None and any(name == "foo" and start <= 4 <= end for name, start, end in ranges)
|
||||
assert ranges is not None and any(name == "foo" and start <= 4 <= end for name, start, end, _ in ranges)
|
||||
assert function_at(ranges, 4) == "foo"
|
||||
|
||||
|
||||
def test_declarator_name_knows_perl_subs_and_never_names_calls() -> None:
|
||||
from openultrasast.semantic.functions import declarator_name
|
||||
from openultrasast.source import declarator_name
|
||||
|
||||
assert declarator_name("sub link_hash_cert {") == "link_hash_cert"
|
||||
assert declarator_name(" return foo(bar)") is None
|
||||
@@ -798,7 +805,7 @@ def test_declarator_name_knows_perl_subs_and_never_names_calls() -> None:
|
||||
|
||||
|
||||
def test_declarator_name_strips_cpp_qualification_to_match_unqualified_labels() -> None:
|
||||
from openultrasast.semantic.functions import declarator_name
|
||||
from openultrasast.source import declarator_name
|
||||
|
||||
assert declarator_name("bool FileReaderLoader::ArrayBufferResult(int a) {") == "ArrayBufferResult"
|
||||
assert declarator_name("void ns::Klass::run(const std::string& s) const {") == "run"
|
||||
@@ -808,7 +815,9 @@ def test_declarator_name_strips_cpp_qualification_to_match_unqualified_labels()
|
||||
@pytest.mark.semantic
|
||||
def test_cpp_qualified_method_range_is_found_by_its_unqualified_label() -> None:
|
||||
from openultrasast.semantic.extra import has_semantic_extra
|
||||
from openultrasast.semantic.functions import function_at, named_function_ranges, spans_named
|
||||
from openultrasast.source import enclosing as function_at
|
||||
from openultrasast.source import ranges as named_function_ranges
|
||||
from openultrasast.source import spans_named
|
||||
|
||||
if not has_semantic_extra():
|
||||
pytest.skip("tree-sitter extra absent")
|
||||
|
||||
@@ -98,13 +98,13 @@ def _produce(world: dict[str, Any], docker: FakeDocker, tmp_path: Path) -> list[
|
||||
|
||||
|
||||
def test_recorded_engine_findings_become_alert_rows() -> None:
|
||||
assert parse_site("app/index.php:3:<global>") == ("app/index.php", 3, "<global>")
|
||||
assert parse_site("app/index.php:3:<module>") == ("app/index.php", 3, "<module>")
|
||||
assert parse_site("lib/A.php:76:Foo::bar") == ("lib/A.php", 76, "Foo::bar")
|
||||
record = read_record(FIXTURE / "probe.json", LOG, 0)
|
||||
assert (record["files"], record["bytes"], record["questions"], record["completed"]) == (1, 145, 1, 1)
|
||||
rows = engine_rows(record, FIXTURE, "vulnerable", "a" * 40, {"app/index.php": [[3, 3]]})
|
||||
assert rows == [
|
||||
{"rule_id": "engine:injection", "rule_status": "enabled", "path": "app/index.php", "line": line, "function": "<global>",
|
||||
{"rule_id": "engine:injection", "rule_status": "enabled", "path": "app/index.php", "line": line, "function": "<module>",
|
||||
"pin_role": "vulnerable", "pin": "a" * 40, "in_fix_range": line == 3, "source": "engine"}
|
||||
for line in (3, 5)
|
||||
] # fmt: skip
|
||||
|
||||
@@ -45,7 +45,7 @@ def _units() -> list[Unit]:
|
||||
return [excerpt, roles_only, oversized, repo]
|
||||
|
||||
|
||||
GLOBAL_ = "<global>"
|
||||
GLOBAL_ = "<module>"
|
||||
|
||||
|
||||
def _render(tmp_path: Path, ceiling: float = 10.0) -> tuple[dict[str, str], Path]:
|
||||
|
||||
@@ -20,7 +20,7 @@ from openultrasast.plane.budget import prices_from
|
||||
from openultrasast.plane.generate import TEMPLATES, fix_ranges, increment, read_cases, render, repin_templates
|
||||
from openultrasast.plane.manifests import load_manifests
|
||||
from openultrasast.plane.memory import MEMORY_KEY_ANNOTATION, POPULATION_ANNOTATION, SPLIT_ANNOTATION, parse_memory_key
|
||||
from openultrasast.plane.tasks.repo_facts import _digest
|
||||
from openultrasast.source import _digest
|
||||
|
||||
PLANE = Path("plane")
|
||||
RUN = PLANE / "runs" / "validation-46.yaml"
|
||||
|
||||
@@ -34,7 +34,7 @@ from openultrasast.plane.memory import (
|
||||
seed,
|
||||
where_sql,
|
||||
)
|
||||
from openultrasast.plane.tasks.repo_facts import _digest
|
||||
from openultrasast.source import _digest
|
||||
|
||||
PIN = "0123456789abcdef0123456789abcdef01234567"
|
||||
OTHER_PIN = "fedcba9876543210fedcba9876543210fedcba98"
|
||||
|
||||
@@ -93,7 +93,7 @@ def test_cross_file_callers_have_line_and_enclosing_function(tree: Path, tmp_pat
|
||||
]
|
||||
assert facts["callers"]["app/util.py::render"] == [
|
||||
{"path": "app/handlers.py", "line": 6, "enclosing": "handler"},
|
||||
{"path": "app/handlers.py", "line": 11, "enclosing": "<global>"},
|
||||
{"path": "app/handlers.py", "line": 11, "enclosing": "<module>"},
|
||||
]
|
||||
assert facts["counts"] == {"files": 3, "functions": 7, "candidates": 2, "callers": 4}
|
||||
summary = json.loads((out / "summary.json").read_text())
|
||||
@@ -271,4 +271,4 @@ def test_enclosing_function_in_python_respects_indentation() -> None:
|
||||
assert repo_facts.enclosing(lines, 1, "python") == "a"
|
||||
assert repo_facts.enclosing(lines, 5, "python") == "inner"
|
||||
assert repo_facts.enclosing(lines, 6, "python") == "b"
|
||||
assert repo_facts.enclosing(lines, 8, "python") == "<global>"
|
||||
assert repo_facts.enclosing(lines, 8, "python") == "<module>"
|
||||
|
||||
@@ -270,10 +270,10 @@ def test_agree_reports_cost_with_the_recorded_triage_and_needs_fix_ranges_to_ass
|
||||
|
||||
|
||||
def test_matches_declared_is_the_protocol_v2_rule() -> None:
|
||||
case = {"family": "injection", "sites": ["a.py::run", "b.py::<global>"]}
|
||||
case = {"family": "injection", "sites": ["a.py::run", "b.py::<module>"]}
|
||||
ranges = {"a.py": [[50, 52]], "c.py": [[5, 5]]}
|
||||
hit = lambda site, family="injection": agree.matches_declared({"site": site, "family": family}, case, ranges) # noqa: E731
|
||||
assert hit("a.py:0:run") and hit("b.py:0:<global>") # a declared path::function, whether or not the fix touched it
|
||||
assert hit("a.py:0:run") and hit("b.py:0:<module>") # a declared path::function, whether or not the fix touched it
|
||||
assert not hit("a.py:0:run", "ssrf") # another family never matches
|
||||
assert hit("a.py:62:helper") and not hit("a.py:63:helper") and hit("a.py:40:x") and not hit("a.py:39:x") # 10-line window
|
||||
assert hit("c.py:900:run") and not hit("d.py:1:run") # a declared function name in a changed file only
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Repository facts use the same names, ranges and calls in every consumer."""
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast import source
|
||||
from openultrasast.semantic import cst
|
||||
from openultrasast.semantic.extra import grammar_for
|
||||
|
||||
SAMPLES = [
|
||||
("python", "app.py", "def run(x):\n return x\n", (1, 2)),
|
||||
("javascript", "app.js", "function run(x) {\n return x;\n}\n", (1, 3)),
|
||||
("java", "App.java", "class App {\n public int run(int x) {\n return x;\n }\n}\n", (2, 4)),
|
||||
("go", "app.go", "package app\nfunc (r *Repo) run(x int) int {\n return x\n}\n", (2, 4)),
|
||||
("ruby", "app.rb", "class App\n def run(x)\n x\n end\nend\n", (2, 4)),
|
||||
("php", "app.php", "<?php\nfunction run($x) {\n return $x;\n}\n", (2, 4)),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("language", "path", "text", "expected"), SAMPLES)
|
||||
@pytest.mark.parametrize("with_grammar", [False, True])
|
||||
def test_ranges_with_and_without_grammars(monkeypatch, language, path, text, expected, with_grammar):
|
||||
if not with_grammar:
|
||||
monkeypatch.setenv("OPENULTRASAST_TREE_SITTER_PROBE", "off")
|
||||
else:
|
||||
monkeypatch.delenv("OPENULTRASAST_TREE_SITTER_PROBE", raising=False)
|
||||
items = source.ranges(path, text, language)
|
||||
assert source.bounds(items, "run") == expected
|
||||
assert source.enclosing(items, expected[0]) == "run"
|
||||
assert source.enclosing(items, expected[1]) == "run"
|
||||
if with_grammar and grammar_for(language) is not None:
|
||||
assert cst.parse_with_cst(path, text, language).engine == "tree-sitter"
|
||||
|
||||
|
||||
def test_nested_ranges_and_module_bounds():
|
||||
text = "def outer():\n def inner():\n pass\n inner()\nouter()\n"
|
||||
items = source.ranges("app.py", text, "python")
|
||||
assert source.enclosing(items, 3) == "inner"
|
||||
assert source.enclosing(items, 4) == "outer"
|
||||
assert source.enclosing(items, 5) == source.MODULE_FUNCTION
|
||||
assert source.enclosing(items, None) == source.MODULE_FUNCTION
|
||||
assert source.bounds(items, "outer") == (1, 4)
|
||||
assert source.bounds(items, source.MODULE_FUNCTION) == (1, 5)
|
||||
assert source.bounds(items, "missing") is None
|
||||
|
||||
|
||||
def test_fallback_does_not_include_siblings_or_strings(monkeypatch):
|
||||
monkeypatch.setenv("OPENULTRASAST_TREE_SITTER_PROBE", "off")
|
||||
items = source.ranges("app.js", 'function first() {\n const s = "}"; // }\n}\nfunction last() {}\nfirst();', "javascript")
|
||||
assert source.bounds(items, "first") == (1, 3)
|
||||
assert source.bounds(items, "last") == (4, 4)
|
||||
assert source.enclosing(items, 5) == source.MODULE_FUNCTION
|
||||
|
||||
|
||||
def test_calls_skip_declarations_comments_strings_and_duplicate_mentions():
|
||||
text = 'def run():\n target(1); target(2)\n# target(3)\ns = "target(4)"\nobj.target(5)\nrun()\n'
|
||||
assert source.call_sites(text, ["target", "run"]) == (
|
||||
source.CallSite(2, "target"),
|
||||
source.CallSite(5, "target"),
|
||||
source.CallSite(6, "run"),
|
||||
)
|
||||
assert source.call_sites(text, []) == ()
|
||||
assert source.call_sites("function run() { return run(); }", ["run"]) == (source.CallSite(1, "run"),)
|
||||
|
||||
|
||||
def test_callers_scan_only_supplied_files(tmp_path):
|
||||
(tmp_path / "app.py").write_text("def run():\n pass\nrun()\n")
|
||||
(tmp_path / "web.py").write_text("def handler():\n run()\n")
|
||||
assert source.callers_of(tmp_path, "run", files=["web.py", "app.py"]) == (("app.py", 3), ("web.py", 2))
|
||||
assert source.callers_of(tmp_path, "run", files=["web.py"]) == (("web.py", 2),)
|
||||
assert source.callers_of(tmp_path, "absent", files=["web.py"]) == ()
|
||||
with pytest.raises(OSError):
|
||||
source.callers_of(tmp_path, "run", files=["missing.py"])
|
||||
with pytest.raises(ValueError):
|
||||
source.callers_of(tmp_path, "run", files=["../outside.py"])
|
||||
|
||||
|
||||
def test_product_files_exclude_vendor_tests_and_private_paths(tmp_path):
|
||||
names = [
|
||||
"src/app.py",
|
||||
"vendor/a.py",
|
||||
"node_modules/a.js",
|
||||
"tests/a.py",
|
||||
"spec/a.rb",
|
||||
"src/a_test.go",
|
||||
"src/AppTest.java",
|
||||
"src/a.test.js",
|
||||
"dist/a.js",
|
||||
".hidden/a.py",
|
||||
".env",
|
||||
"src/a.min.js",
|
||||
]
|
||||
for name in names:
|
||||
path = tmp_path / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("run()\n")
|
||||
(tmp_path / "link.py").symlink_to(tmp_path / "src/app.py")
|
||||
assert source.product_files(tmp_path) == ["src/app.py"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("name", "expected"),
|
||||
[
|
||||
("py", "python"),
|
||||
("python_web", "python"),
|
||||
("node", "javascript"),
|
||||
("js", "javascript"),
|
||||
("ts", "typescript"),
|
||||
("typescript", "typescript"),
|
||||
("c_cpp", "cpp"),
|
||||
("c++", "cpp"),
|
||||
("cpp", "cpp"),
|
||||
("Ruby", "ruby"),
|
||||
("unknown", "unknown"),
|
||||
],
|
||||
)
|
||||
def test_canonical_language(name, expected):
|
||||
assert source.canonical_language(name) == expected
|
||||
|
||||
|
||||
def test_facts_reuse_supplied_entry_points_and_read_only_repository(tmp_path, monkeypatch):
|
||||
from openultrasast import mapping
|
||||
|
||||
(tmp_path / "app.py").write_text("def run(x):\n return x\n")
|
||||
(tmp_path / "web.py").write_text("def handler():\n run(1)\n")
|
||||
monkeypatch.setattr(mapping, "analyze_entry_points", lambda *_: pytest.fail("entry points were recomputed"))
|
||||
result = source.facts_for(tmp_path, "app.py", 2, entry_points=[])
|
||||
assert result == source.SourceFacts("python", "run", (1, 2), (("web.py", 2),), ())
|
||||
|
||||
|
||||
def test_parse_rejection_is_not_an_empty_result(monkeypatch):
|
||||
from openultrasast.semantic.ir import FileIR
|
||||
|
||||
monkeypatch.setattr(
|
||||
cst, "parse_with_cst", lambda path, text, language: FileIR(path, language, "tree-sitter", (), False, "parse_failed")
|
||||
)
|
||||
with pytest.raises(ValueError, match="source parse failed"):
|
||||
source.ranges("app.py", "def broken(", "python")
|
||||
Reference in New Issue
Block a user