diff --git a/.kiro/breadth.toml b/.kiro/breadth.toml index 4f8bcdf..3ff5293 100644 --- a/.kiro/breadth.toml +++ b/.kiro/breadth.toml @@ -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] diff --git a/benchmarks/independent/freeze.py b/benchmarks/independent/freeze.py index 876646c..28329b6 100644 --- a/benchmarks/independent/freeze.py +++ b/benchmarks/independent/freeze.py @@ -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 ``, which the freeze record shows.""" + that defeats it yields no function, and the site falls back to ``, 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 ``. For a pure insertion, `line` is the line the + """The innermost named function containing `line`, or ``. 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 "" + return min(inside)[1] if inside else "" 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 ``) of `path` at the vulnerable pin that the fix's hunks touch, in hunk order.""" + """The functions (or ``) 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 == ""): + if name not in names and not (inserted and name == ""): 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 ``)?""" + """Per declared site: does the fix's diff change a line inside that function (or the file, for ``)?""" result = {} for site in sites: path, _, name = site.partition("::") hunks = old_hunks(repo, str(case["vulnerable"]), str(case["fixed"]), path) - if name == "" or not hunks: + if name == "" 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("::")] + functions = [site.partition("::")[2] for site in sites if site.partition("::")[0] == sink_file and not site.endswith("::")] else: sink_file, functions = sink_parts(str(case["sink"])) record["sink_file"], record["sink_functions"] = sink_file, functions diff --git a/benchmarks/learn/build_features.py b/benchmarks/learn/build_features.py index fdc87eb..90ae285 100644 --- a/benchmarks/learn/build_features.py +++ b/benchmarks/learn/build_features.py @@ -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: diff --git a/benchmarks/measurements/2026-10-13-source-module-label-diff/compare_labels.py b/benchmarks/measurements/2026-10-13-source-module-label-diff/compare_labels.py new file mode 100644 index 0000000..7b867c0 --- /dev/null +++ b/benchmarks/measurements/2026-10-13-source-module-label-diff/compare_labels.py @@ -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("", "") + 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") diff --git a/benchmarks/measurements/2026-10-13-source-module-label-diff/record.json b/benchmarks/measurements/2026-10-13-source-module-label-diff/record.json new file mode 100644 index 0000000..c2b0634 --- /dev/null +++ b/benchmarks/measurements/2026-10-13-source-module-label-diff/record.json @@ -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 to .", + "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::; 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/." +} \ No newline at end of file diff --git a/benchmarks/pairs/advisory-fixes/materialize.py b/benchmarks/pairs/advisory-fixes/materialize.py index e380cca..05f1b0d 100644 --- a/benchmarks/pairs/advisory-fixes/materialize.py +++ b/benchmarks/pairs/advisory-fixes/materialize.py @@ -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(): diff --git a/benchmarks/pairs/catalog_gen.py b/benchmarks/pairs/catalog_gen.py index c5474d1..b863a3a 100644 --- a/benchmarks/pairs/catalog_gen.py +++ b/benchmarks/pairs/catalog_gen.py @@ -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) diff --git a/src/openultrasast/hunter_tools.py b/src/openultrasast/hunter_tools.py index 195346f..1e2efa4 100644 --- a/src/openultrasast/hunter_tools.py +++ b/src/openultrasast/hunter_tools.py @@ -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]]: diff --git a/src/openultrasast/learn/excerpt.py b/src/openultrasast/learn/excerpt.py index d788a15..cf597d4 100644 --- a/src/openultrasast/learn/excerpt.py +++ b/src/openultrasast/learn/excerpt.py @@ -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 = "" -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 ````, 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"(?$]){quoted}\s*(?:<[^<>(){{}};]*>\s*)?(?=\()" - rf"|^\s*sub\s+{quoted}\s*(?=[{{(;])" - rf"|(? 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 ````), 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 diff --git a/src/openultrasast/learn/features.py b/src/openultrasast/learn/features.py index aba7352..07d790f 100644 --- a/src/openultrasast/learn/features.py +++ b/src/openultrasast/learn/features.py @@ -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. ```` 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"(? 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",)) diff --git a/src/openultrasast/model/candidates.py b/src/openultrasast/model/candidates.py index 71c022c..ba9ba95 100644 --- a/src/openultrasast/model/candidates.py +++ b/src/openultrasast/model/candidates.py @@ -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) diff --git a/src/openultrasast/pairs.py b/src/openultrasast/pairs.py index 23f41a0..b02eec8 100644 --- a/src/openultrasast/pairs.py +++ b/src/openultrasast/pairs.py @@ -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); "" 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 diff --git a/src/openultrasast/plane/harvest.py b/src/openultrasast/plane/harvest.py index 70354b6..d357c8b 100644 --- a/src/openultrasast/plane/harvest.py +++ b/src/openultrasast/plane/harvest.py @@ -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 ```` or when not declared.""" + """1-based line of the first declaration of ``function`` in ``lines``; 1 for ```` 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 diff --git a/src/openultrasast/plane/tasks/alerts.py b/src/openultrasast/plane/tasks/alerts.py index 5af6923..123fb74 100644 --- a/src/openultrasast/plane/tasks/alerts.py +++ b/src/openultrasast/plane/tasks/alerts.py @@ -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``; ```` outside any function); a language those patterns do not cover keeps the +candidates' ``path::function``; ```` 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`` (````).""" + """The enclosing declaration of ``path:line`` by the ``repo-facts`` patterns, else ``fallback`` (````).""" 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]]: diff --git a/src/openultrasast/plane/tasks/features.py b/src/openultrasast/plane/tasks/features.py index 0c5c5be..f2769a8 100644 --- a/src/openultrasast/plane/tasks/features.py +++ b/src/openultrasast/plane/tasks/features.py @@ -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 diff --git a/src/openultrasast/plane/tasks/repo_facts.py b/src/openultrasast/plane/tasks/repo_facts.py index c758054..80d8aec 100644 --- a/src/openultrasast/plane/tasks/repo_facts.py +++ b/src/openultrasast/plane/tasks/repo_facts.py @@ -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 = "" -_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), `` 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"(? 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"]))): diff --git a/src/openultrasast/plane/tasks/roles.py b/src/openultrasast/plane/tasks/roles.py index a06c97f..1eb8ee6 100644 --- a/src/openultrasast/plane/tasks/roles.py +++ b/src/openultrasast/plane/tasks/roles.py @@ -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": "", "function": ">", ' + '{{"roles": [{{"role": "", "function": ">", ' '"line": , "call": "", "operation": "", ' '"evidence": ""}}]}}. 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", "") diff --git a/src/openultrasast/semantic/facts.py b/src/openultrasast/semantic/facts.py index f7117c4..6806f27 100644 --- a/src/openultrasast/semantic/facts.py +++ b/src/openultrasast/semantic/facts.py @@ -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), diff --git a/src/openultrasast/semantic/functions.py b/src/openultrasast/semantic/functions.py index 0252d05..2ba6c43 100644 --- a/src/openultrasast/semantic/functions.py +++ b/src/openultrasast/semantic/functions.py @@ -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 ````. 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 = "" -ANON_FUNCTION = "" -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 ```` 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|(? 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) diff --git a/src/openultrasast/semantic/obligations/facts.py b/src/openultrasast/semantic/obligations/facts.py index 4762db8..38c016a 100644 --- a/src/openultrasast/semantic/obligations/facts.py +++ b/src/openultrasast/semantic/obligations/facts.py @@ -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), diff --git a/src/openultrasast/semantic/obligations/shapes.py b/src/openultrasast/semantic/obligations/shapes.py index ffc8c1b..4c96bfd 100644 --- a/src/openultrasast/semantic/obligations/shapes.py +++ b/src/openultrasast/semantic/obligations/shapes.py @@ -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: diff --git a/src/openultrasast/semantic/overlay.py b/src/openultrasast/semantic/overlay.py index 05c94c3..8cd8fef 100644 --- a/src/openultrasast/semantic/overlay.py +++ b/src/openultrasast/semantic/overlay.py @@ -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) diff --git a/src/openultrasast/source.py b/src/openultrasast/source.py new file mode 100644 index 0000000..7bba6d0 --- /dev/null +++ b/src/openultrasast/source.py @@ -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 = "" +LEGACY_MODULE_FUNCTION = "" # what frozen records before 2026-10-13 call module-level code +ANON_FUNCTION = "" +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|(? 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"(?])([A-Za-z_$][\w:$]*)\s*(?:<[^<>(){};]*>\s*)?(?=\()|^\s*sub\s+(\w+)\s*(?=\{)" + r"|(?)" +) + + +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"(?" 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"(?$]){quoted}\s*(?:<[^<>(){{}};]*>\s*)?(?=\()" + rf"|^\s*sub\s+{quoted}\s*(?=[{{(;])" + rf"|(? 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 ````), 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), `` 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"(? 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. ```` 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 diff --git a/tests/fixtures/plane-engine/probe.json b/tests/fixtures/plane-engine/probe.json index 8c770f9..c499a44 100644 --- a/tests/fixtures/plane-engine/probe.json +++ b/tests/fixtures/plane-engine/probe.json @@ -12,14 +12,14 @@ "degradations": [], "findings": [ { - "site": "app/index.php:3:", + "site": "app/index.php:3:", "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:", + "site": "app/index.php:5:", "family": "injection", "rung": "model_entailed", "witness": "$_GET[\"host\"] -> system(\"ping \" . $_GET[\"host\"]) in app/index.php (line 5, 2 steps)", diff --git a/tests/test_hunter_tools.py b/tests/test_hunter_tools.py index f49d833..1f520b4 100644 --- a/tests/test_hunter_tools.py +++ b/tests/test_hunter_tools.py @@ -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, "") == [] diff --git a/tests/test_independent_freeze.py b/tests/test_independent_freeze.py index daa132e..9787e3a 100644 --- a/tests/test_independent_freeze.py +++ b/tests/test_independent_freeze.py @@ -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 ``.""" +outside every function is ``.""" 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) == "" - assert freeze.enclosing(spans, 1) == "" + assert freeze.enclosing(spans, 21) == "" + assert freeze.enclosing(spans, 1) == "" 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) == "" # new code that follows it + assert freeze.enclosing(spans, 10, inserted=True) == "" # new code that follows it assert freeze.enclosing(spans, 8, inserted=True) == "open" diff --git a/tests/test_pair_corpus.py b/tests/test_pair_corpus.py index cf932ec..6321763 100644 --- a/tests/test_pair_corpus.py +++ b/tests/test_pair_corpus.py @@ -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 = (("", 1, 40), ("handler", 3, 20), ("", 5, 12)) + ranges = ( + FunctionRange("", 1, 40, "module"), + FunctionRange("handler", 3, 20, "function"), + FunctionRange("", 5, 12, "anonymous"), + ) assert function_at(ranges, 7) == "handler" # innermost *named* function, never "" - assert function_at(ranges, 25) is None - assert function_at((("", 1, 9),), 4) is None + assert function_at(ranges, 25) == "" + assert function_at((FunctionRange("", 1, 9, "anonymous"),), 4) == "" @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 "" not in {name for name, _, _ in ranges} + assert {name for name, _, _, _ in ranges} >= {"a", "b", "c", "d", "e"} + assert "" 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") diff --git a/tests/test_plane_engine_alerts.py b/tests/test_plane_engine_alerts.py index d8c051c..372837b 100644 --- a/tests/test_plane_engine_alerts.py +++ b/tests/test_plane_engine_alerts.py @@ -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:") == ("app/index.php", 3, "") + assert parse_site("app/index.php:3:") == ("app/index.php", 3, "") 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": "", + {"rule_id": "engine:injection", "rule_status": "enabled", "path": "app/index.php", "line": line, "function": "", "pin_role": "vulnerable", "pin": "a" * 40, "in_fix_range": line == 3, "source": "engine"} for line in (3, 5) ] # fmt: skip diff --git a/tests/test_plane_harvest.py b/tests/test_plane_harvest.py index 80338d0..c3bb8ec 100644 --- a/tests/test_plane_harvest.py +++ b/tests/test_plane_harvest.py @@ -45,7 +45,7 @@ def _units() -> list[Unit]: return [excerpt, roles_only, oversized, repo] -GLOBAL_ = "" +GLOBAL_ = "" def _render(tmp_path: Path, ceiling: float = 10.0) -> tuple[dict[str, str], Path]: diff --git a/tests/test_plane_increment.py b/tests/test_plane_increment.py index 613b688..87f868c 100644 --- a/tests/test_plane_increment.py +++ b/tests/test_plane_increment.py @@ -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" diff --git a/tests/test_plane_memory.py b/tests/test_plane_memory.py index 8e8d002..e58cb58 100644 --- a/tests/test_plane_memory.py +++ b/tests/test_plane_memory.py @@ -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" diff --git a/tests/test_plane_repo_facts.py b/tests/test_plane_repo_facts.py index 981dbc4..529089f 100644 --- a/tests/test_plane_repo_facts.py +++ b/tests/test_plane_repo_facts.py @@ -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": ""}, + {"path": "app/handlers.py", "line": 11, "enclosing": ""}, ] 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") == "" + assert repo_facts.enclosing(lines, 8, "python") == "" diff --git a/tests/test_plane_verify.py b/tests/test_plane_verify.py index f1f8d98..69cb1bc 100644 --- a/tests/test_plane_verify.py +++ b/tests/test_plane_verify.py @@ -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::"]} + case = {"family": "injection", "sites": ["a.py::run", "b.py::"]} 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:") # a declared path::function, whether or not the fix touched it + assert hit("a.py:0:run") and hit("b.py:0:") # 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 diff --git a/tests/test_source.py b/tests/test_source.py new file mode 100644 index 0000000..c9abafb --- /dev/null +++ b/tests/test_source.py @@ -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", "