diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 375129c..3778b0e 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -54,7 +54,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Boundary: TestTuningAdvisor_ - _Depends: 2.2_ -- [ ] 2.4 Apply a complexity ledger overlay that cannot hide severity-5 reachable inventory +- [x] 2.4 Apply a complexity ledger overlay that cannot hide severity-5 reachable inventory - Triggerable verdicts raise a hotspot's later score; not-triggerable verdicts lower it; the underlying inventory hit remains. - A reachable policy-severity-5 inventory finding stays in the regression candidate set regardless of a negative ledger delta. - Observable completion: after a not-triggerable overlay the hotspot score drops, and a sev-5 reachable finding is still selected as a candidate, proven by a unit test. @@ -63,7 +63,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Depends: 2.2_ - [ ] 3. Add repo-bound hunter tools and a scripted tool loop -- [ ] 3.1 (P) Clamp tool paths and expose read, grep, and reference search +- [x] 3.1 (P) Clamp tool paths and expose read, grep, and reference search - Read, grep, and reference listing resolve only under the scan root; `..` and absolute escapes raise a path-escape error. - Grep runs with the standard library over enumerated source files. - Observable completion: a unit test reads a repo file, greps a split-sink assignment, and rejects `../etc/passwd`. diff --git a/src/openultrasast/hunter_tools.py b/src/openultrasast/hunter_tools.py new file mode 100644 index 0000000..183f96c --- /dev/null +++ b/src/openultrasast/hunter_tools.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import os +import re +from collections.abc import Iterable, Iterator, Mapping +from pathlib import Path + +from .preprocess import enumerate_source_files + + +class PathEscapesRepo(ValueError): + """Raised when a resolved tool path is not under the scan root.""" + + +def clamp_repo_path(root: Path, user_path: str) -> Path: + resolved_root = root.resolve() + candidate = Path(user_path) + resolved = candidate.resolve() if candidate.is_absolute() else (resolved_root / candidate).resolve() + try: + common = os.path.commonpath([str(resolved_root), str(resolved)]) + except ValueError as exc: + raise PathEscapesRepo(f"path {user_path!r} escapes scan root {resolved_root}") from exc + if common != str(resolved_root): + raise PathEscapesRepo(f"path {user_path!r} escapes scan root {resolved_root}") + return resolved + + +def read_file(root: Path, path: str, *, max_chars: int) -> str: + clamped = clamp_repo_path(root, path) + if max_chars <= 0: + return "" + return clamped.read_text(errors="ignore")[:max_chars] + + +def grep_repo(root: Path, pattern: str, *, max_matches: int) -> list[dict[str, object]]: + if max_matches <= 0: + return [] + compiled = re.compile(pattern) + matches: list[dict[str, object]] = [] + for relative, path in _iter_clamped_source_files(root): + for line_no, line in enumerate(path.read_text(errors="ignore").splitlines(), start=1): + if compiled.search(line): + matches.append({"path": relative, "line": line_no, "text": line}) + if len(matches) >= max_matches: + return matches + return matches + + +def find_refs(root: Path, symbol: str, mapping_index: object) -> list[dict[str, object]]: + resolved_root = root.resolve() + refs: list[dict[str, object]] = [] + for record in _iter_mapping_records(mapping_index): + if not _record_matches_symbol(record, symbol): + continue + raw_path = _record_path(record) + if raw_path is None: + continue + clamped = clamp_repo_path(resolved_root, raw_path) + name = _record_value(record, "name", "function_name", "symbol") + refs.append( + { + "path": clamped.relative_to(resolved_root).as_posix(), + "name": str(name) if name is not None else symbol, + "line": _as_line(_record_value(record, "line")), + "source": "mapping", + } + ) + if symbol: + compiled = re.compile(rf"\b{re.escape(symbol)}\b") + for relative, path in _iter_clamped_source_files(resolved_root): + for line_no, line in enumerate(path.read_text(errors="ignore").splitlines(), start=1): + if compiled.search(line): + refs.append({"path": relative, "name": symbol, "line": line_no, "text": line, "source": "text"}) + return refs + + +def _iter_clamped_source_files(root: Path) -> Iterator[tuple[str, Path]]: + resolved_root = root.resolve() + for path in enumerate_source_files(resolved_root): + relative = path.relative_to(resolved_root).as_posix() + yield relative, clamp_repo_path(resolved_root, relative) + + +def _iter_mapping_records(mapping_index: object) -> Iterable[object]: + if mapping_index is None: + return () + if isinstance(mapping_index, Mapping): + return (mapping_index,) + if isinstance(mapping_index, (str, bytes, bytearray)): + return () + if isinstance(mapping_index, Iterable): + return mapping_index + return (mapping_index,) + + +def _record_matches_symbol(record: object, symbol: str) -> bool: + for field in ("name", "function_name", "symbol"): + value = _record_value(record, field) + if value is not None and str(value) == symbol: + return True + return False + + +def _record_path(record: object) -> str | None: + value = record.get("path") if isinstance(record, Mapping) else getattr(record, "path", None) + if value is None: + return None + return str(value) + + +def _record_value(record: object, *names: str) -> object | None: + for name in names: + raw: object | None + if isinstance(record, Mapping): + if name not in record: + continue + raw = record[name] + else: + raw = getattr(record, name, None) + if raw is not None: + return raw + return None + + +def _as_line(value: object) -> int | None: + if isinstance(value, bool) or value is None: + return None + if isinstance(value, int): + return value + return None diff --git a/tests/test_hunter_tools.py b/tests/test_hunter_tools.py new file mode 100644 index 0000000..72c9589 --- /dev/null +++ b/tests/test_hunter_tools.py @@ -0,0 +1,123 @@ +from pathlib import Path + +import pytest + +from openultrasast.hunter_tools import PathEscapesRepo, clamp_repo_path, find_refs, grep_repo, read_file +from openultrasast.mapping import EntryPointRecord + +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_uses_mapping_index_and_clamped_text_search(tmp_path: Path) -> None: + root = _repo_with_split_sink(tmp_path) + mapping_index = [ + {"path": "app.py", "name": "query", "line": 2}, + EntryPointRecord( + path="app.py", + line=3, + end_line=3, + function_name="search", + name="search", + kind="route", + access_level="public", + trust_boundary="http_request", + access_evidence=[], + conditions=[], + provenance="test", + rationale="fixture", + ), + ] + + query_refs = find_refs(root, "query", mapping_index) + assert any(ref["path"] == "app.py" and ref["name"] == "query" for ref in query_refs) + assert any(ref.get("source") == "mapping" for ref in query_refs) + assert any(ref.get("source") == "text" and "query" in str(ref.get("text", ref.get("name", ""))) for ref in query_refs) + + search_refs = find_refs(root, "search", mapping_index) + assert any(ref["path"] == "app.py" and ref["name"] == "search" for ref in search_refs) + + +def test_find_refs_rejects_mapping_path_escape(tmp_path: Path) -> None: + root = _repo_with_split_sink(tmp_path) + + with pytest.raises(PathEscapesRepo): + find_refs(root, "passwd", [{"path": "../etc/passwd", "name": "passwd"}]) + with pytest.raises(PathEscapesRepo): + find_refs(root, "passwd", [{"path": "/etc/passwd", "name": "passwd"}])