mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Fable 5.1
062a7b780e
src/openultrasast/source.py replaces the copies in semantic/functions.py, plane/tasks/repo_facts.py, learn/features.py, learn/excerpt.py and plane/tasks/verify.py. Two engines share one sentinel (<module>), one DECLARATION/KEYWORDS/RESERVED set and one alias map: the parser engine (ranges, enclosing, bounds, call_sites, callers_of, facts_for) for the scorer, the overlay and the hunter's annotations; the line-pattern engine (declared_functions, enclosing_at, function_span, function_bounds, the brace matcher) that labels, features, roles, excerpts and plane facts were built on. learn no longer imports plane.tasks. Frozen records that store the old sentinel are mapped on read (module_function). Gate: labels rebuilt before and after are identical, 7,921 rows, after sentinel normalisation (benchmarks/measurements/2026-10-13-source-module- label-diff/record.json). 490 targeted tests pass; 3,312 collect. Net: 657 lines deleted, 212 added. benchmarks/unseen keeps <global> for its own frozen manifests. Co-Authored-By: Claude Fable 5.1 <[email protected]>
128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from openultrasast.hunter_tools import PathEscapesRepo, clamp_repo_path, find_refs, grep_repo, read_file
|
|
|
|
SPLIT_SINK = """\
|
|
term = request.args["q"]
|
|
query = "select * from items where title like '%" + term + "%'"
|
|
return db.execute(query)
|
|
"""
|
|
|
|
|
|
def _repo_with_split_sink(tmp_path: Path) -> Path:
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
(root / "app.py").write_text(SPLIT_SINK)
|
|
return root
|
|
|
|
|
|
def test_read_grep_and_reject_path_escape(tmp_path: Path) -> None:
|
|
root = _repo_with_split_sink(tmp_path)
|
|
|
|
source = read_file(root, "app.py", max_chars=4000)
|
|
assert "query =" in source
|
|
assert "db.execute(query)" in source
|
|
|
|
matches = grep_repo(root, r"query\s*=", max_matches=10)
|
|
assert matches
|
|
assert matches[0]["path"] == "app.py"
|
|
assert matches[0]["line"] == 2
|
|
assert "query =" in str(matches[0]["text"])
|
|
|
|
with pytest.raises(PathEscapesRepo):
|
|
read_file(root, "../etc/passwd", max_chars=100)
|
|
|
|
|
|
def test_clamp_repo_path_rejects_dotdot_and_absolute_escapes(tmp_path: Path) -> None:
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
(root / "app.py").write_text("ok\n")
|
|
|
|
assert clamp_repo_path(root, "app.py") == (root / "app.py").resolve()
|
|
|
|
with pytest.raises(PathEscapesRepo):
|
|
clamp_repo_path(root, "../etc/passwd")
|
|
with pytest.raises(PathEscapesRepo):
|
|
clamp_repo_path(root, "/etc/passwd")
|
|
with pytest.raises(PathEscapesRepo):
|
|
clamp_repo_path(root, "..")
|
|
|
|
|
|
def test_grep_repo_does_not_search_outside_root(tmp_path: Path) -> None:
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
(root / "app.py").write_text('query = "inside"\n')
|
|
(tmp_path / "outside.py").write_text('query = "outside"\n')
|
|
|
|
matches = grep_repo(root, r"query\s*=", max_matches=10)
|
|
|
|
assert [match["path"] for match in matches] == ["app.py"]
|
|
assert all("outside" not in str(match["text"]) for match in matches)
|
|
|
|
|
|
def test_grep_repo_honors_max_matches(tmp_path: Path) -> None:
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
(root / "app.py").write_text("sink = 1\nsink = 2\nsink = 3\n")
|
|
|
|
matches = grep_repo(root, r"sink\s*=", max_matches=2)
|
|
|
|
assert len(matches) == 2
|
|
assert [match["line"] for match in matches] == [1, 2]
|
|
|
|
|
|
def test_read_file_truncates_and_rejects_symlink_escape(tmp_path: Path) -> None:
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
(root / "app.py").write_text("abcdef")
|
|
(root / "link.py").symlink_to("/etc/passwd")
|
|
|
|
assert read_file(root, "app.py", max_chars=3) == "abc"
|
|
with pytest.raises(PathEscapesRepo):
|
|
read_file(root, "link.py", max_chars=100)
|
|
|
|
|
|
def test_find_refs_searches_source_text(tmp_path: Path) -> None:
|
|
root = _repo_with_split_sink(tmp_path)
|
|
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, "") == []
|
|
|
|
|
|
def test_grep_reads_the_tree_once_and_sees_an_edit(tmp_path, monkeypatch) -> None:
|
|
"""Every call re-enumerated and re-read the repository; now once per root state, re-validated per file."""
|
|
import os
|
|
|
|
from openultrasast import hunter_tools
|
|
|
|
sub = tmp_path / "pkg"
|
|
sub.mkdir()
|
|
source = sub / "app.py"
|
|
source.write_text("def run(q):\n return query(q)\n")
|
|
calls = []
|
|
real = hunter_tools._iter_clamped_source_files
|
|
monkeypatch.setattr(hunter_tools, "_iter_clamped_source_files", lambda root: calls.append(root) or real(root))
|
|
assert [m["line"] for m in hunter_tools.grep_repo(tmp_path, r"query\(", max_matches=5)] == [2]
|
|
assert [m["line"] for m in hunter_tools.grep_repo(tmp_path, r"def run", max_matches=5)] == [1]
|
|
assert len(calls) == 1, "the tree was enumerated again for an unchanged root"
|
|
source.write_text("def run(q):\n x = 1\n return query(q)\n")
|
|
stamp = source.stat().st_mtime_ns + 10_000_000
|
|
os.utime(source, ns=(stamp, stamp))
|
|
assert [m["line"] for m in hunter_tools.grep_repo(tmp_path, r"query\(", max_matches=5)] == [3], "a stale read survived an edit"
|
|
|
|
|
|
def test_a_match_in_a_minified_line_is_quoted_not_returned_whole(tmp_path) -> None:
|
|
"""FUXA's minified bundles have megabyte lines; one match overflowed the model's context for the whole hunt."""
|
|
from openultrasast import hunter_tools, tool_hunter
|
|
|
|
(tmp_path / "bundle.js").write_text("var a=1;" + "x" * 500_000 + ";eval(q);\n")
|
|
[match] = hunter_tools.grep_repo(tmp_path, r"var a", max_matches=5)
|
|
assert len(match["text"]) <= 240
|
|
call = tool_hunter.ToolCall(id="c", name="grep_repo", arguments={"pattern": "x+", "max_matches": 50})
|
|
assert len(tool_hunter._run_tool(tmp_path, call, 4000)) <= 16 * 4000 + 64
|