mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-17 12:09:27 +00:00
bench(independent): freeze population v3 (PHP)
freeze.py, extended for v3 and still reading only git history, file lists and fix diffs: - a file the advisory names without a function resolves to the function(s) the fix's diff changes at the vulnerable pin (maintainer's decision 2), recorded with its basis (advisory-named / resolved-from-fix-diff); a new helper inserted at file scope adds no site; - an inferred fix must change a declared site function, else the case drops; - a single-language population gets a strict benign proposal (PHP change, merges judged with all they bring in, pre-vulnerable history as fallback). All 15 cases verified, none dropped (pagible and podlove fixes confirmed). Nine sites in eight cases resolved from diffs; five benign proposals replaced by hand. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
edcf145b00
commit
0a04dbfada
@@ -78,22 +78,179 @@ def sink_parts(sink: str) -> tuple[str, list[str]]:
|
||||
return path.strip(), functions
|
||||
|
||||
|
||||
def benign(repo: Path, fixed: str, sink_file: str, extensions: tuple[str, ...]) -> dict[str, object] | None:
|
||||
PHP_FUNCTION = re.compile(r"\bfunction\s*&?\s*([A-Za-z_][A-Za-z0-9_]*)\s*\(")
|
||||
|
||||
|
||||
def php_code(text: str) -> str:
|
||||
"""`text` with comments, string contents and inline HTML blanked to spaces, newlines kept (line numbers hold).
|
||||
|
||||
Enough to find function bodies by brace matching; heredoc/nowdoc bodies are blanked too. Not a parser: a file
|
||||
that defeats it yields no function, and the site falls back to `<global>`, which the freeze record shows."""
|
||||
out, i, n, php = list(text), 0, len(text), False
|
||||
|
||||
def blank(start: int, end: int) -> None:
|
||||
for k in range(start, min(end, n)):
|
||||
if out[k] != "\n":
|
||||
out[k] = " "
|
||||
|
||||
while i < n:
|
||||
if not php:
|
||||
j = text.find("<?", i)
|
||||
blank(i, n if j < 0 else j)
|
||||
if j < 0:
|
||||
break
|
||||
php, i = True, j + (5 if text.startswith("<?php", j) else 2)
|
||||
continue
|
||||
if text.startswith("?>", i):
|
||||
php, i = False, i + 2
|
||||
elif text.startswith("//", i) or text[i] == "#" and not text.startswith("#[", i):
|
||||
j = min(k for k in (text.find("\n", i), text.find("?>", i), n) if k >= 0)
|
||||
blank(i, j)
|
||||
i = j
|
||||
elif text.startswith("/*", i):
|
||||
j = text.find("*/", i + 2)
|
||||
j = n if j < 0 else j + 2
|
||||
blank(i, j)
|
||||
i = j
|
||||
elif text.startswith("<<<", i):
|
||||
label = re.match(r"<<<\s*['\"]?([A-Za-z_][A-Za-z0-9_]*)", text[i:])
|
||||
end = re.compile(r"^\s*" + re.escape(label.group(1)) + r"\b", re.M) if label else None
|
||||
found = end.search(text, text.find("\n", i) + 1) if end else None
|
||||
j = found.end() if found else i + 3
|
||||
blank(i, j)
|
||||
i = j
|
||||
elif text[i] in "'\"":
|
||||
quote, j = text[i], i + 1
|
||||
while j < n and text[j] != quote:
|
||||
j += 2 if text[j] == "\\" else 1
|
||||
blank(i + 1, j)
|
||||
i = j + 1
|
||||
else:
|
||||
i += 1
|
||||
return "".join(out)
|
||||
|
||||
|
||||
def php_functions(text: str) -> list[tuple[str, int, int]]:
|
||||
"""(name, first line, last line) of every named function or method body in a PHP file, 1-based."""
|
||||
code, spans = php_code(text), []
|
||||
for match in PHP_FUNCTION.finditer(code):
|
||||
brace, semi = code.find("{", match.end()), code.find(";", match.end())
|
||||
if brace < 0 or 0 <= semi < brace: # abstract or interface method: no body
|
||||
continue
|
||||
depth, k = 0, brace
|
||||
while k < len(code):
|
||||
depth += {"{": 1, "}": -1}.get(code[k], 0)
|
||||
if depth == 0:
|
||||
break
|
||||
k += 1
|
||||
spans.append((match.group(1), code.count("\n", 0, match.start()) + 1, code.count("\n", 0, k) + 1))
|
||||
return spans
|
||||
|
||||
|
||||
def enclosing(spans: list[tuple[str, int, int]], line: int, inserted: bool = False) -> str:
|
||||
"""The innermost named function containing `line`, or `<global>`. For a pure insertion, `line` is the line the
|
||||
new code follows, so a function whose closing brace is on that line does not contain it."""
|
||||
inside = [(last - first, name) for name, first, last in spans if first <= line <= last and not (inserted and line == last)]
|
||||
return min(inside)[1] if inside else "<global>"
|
||||
|
||||
|
||||
def old_hunks(repo: Path, vulnerable: str, fixed: str, path: str) -> list[tuple[int, int, bool]]:
|
||||
"""Vulnerable-side line ranges the fix changes in `path`, and whether each is a pure insertion (then the range
|
||||
is the line it follows, at least 1)."""
|
||||
diff = git(repo, "diff", "-U0", vulnerable, fixed, "--", path)
|
||||
ranges = []
|
||||
for start, count in re.findall(r"^@@ -(\d+)(?:,(\d+))? \+", diff, re.M):
|
||||
first, size = int(start), 1 if count == "" else int(count)
|
||||
ranges.append((max(first, 1), max(first, 1), True) if size == 0 else (first, first + size - 1, False))
|
||||
return ranges
|
||||
|
||||
|
||||
def changed_functions(repo: Path, vulnerable: str, fixed: str, path: str) -> list[str]:
|
||||
"""The functions (or `<global>`) of `path` at the vulnerable pin that the fix's hunks touch, in hunk order."""
|
||||
text = (blob(repo, vulnerable, path) or b"").decode(errors="replace")
|
||||
spans, names = php_functions(text), []
|
||||
for first, last, inserted in old_hunks(repo, vulnerable, fixed, path):
|
||||
for line in range(first, last + 1):
|
||||
name = enclosing(spans, line, inserted)
|
||||
# New code added between functions (a new helper) changes no vulnerable-side code: no site.
|
||||
if name not in names and not (inserted and name == "<global>"):
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
|
||||
def resolve_sites(repo: Path, case: dict[str, object], changed: list[str]) -> dict[str, object]:
|
||||
"""Decision 2 of population v3: a file the advisory names without a function gets, as its declared site, the
|
||||
function(s) the fix's diff of that file changes. Advisory-named sites are kept and marked as such."""
|
||||
declared = [str(s) for s in case.get("sites", [])] # type: ignore[union-attr]
|
||||
recorded = case.get("site_basis") or {} # a frozen population keeps each site's basis
|
||||
basis = {site: recorded.get(site, "advisory-named") for site in declared} # type: ignore[union-attr]
|
||||
unresolved = {}
|
||||
named_files = {site.partition("::")[0] for site in declared}
|
||||
for path in [str(f) for f in case.get("site_files", [])]: # type: ignore[union-attr]
|
||||
if path in named_files:
|
||||
continue
|
||||
if path not in changed:
|
||||
unresolved[path] = "the fix does not change this file"
|
||||
continue
|
||||
for name in changed_functions(repo, str(case["vulnerable"]), str(case["fixed"]), path):
|
||||
basis.setdefault(f"{path}::{name}", "resolved-from-fix-diff")
|
||||
return {"sites": list(basis), "site_basis": basis, "site_files_unresolved": unresolved}
|
||||
|
||||
|
||||
def fix_changes_sites(repo: Path, case: dict[str, object], sites: list[str]) -> dict[str, bool]:
|
||||
"""Per declared site: does the fix's diff change a line inside that function (or the file, for `<global>`)?"""
|
||||
result = {}
|
||||
for site in sites:
|
||||
path, _, name = site.partition("::")
|
||||
hunks = old_hunks(repo, str(case["vulnerable"]), str(case["fixed"]), path)
|
||||
if name == "<global>" or not hunks:
|
||||
result[site] = bool(hunks)
|
||||
continue
|
||||
spans = php_functions((blob(repo, str(case["vulnerable"]), path) or b"").decode(errors="replace"))
|
||||
result[site] = any(enclosing(spans, line, inserted) == name for first, last, inserted in hunks for line in range(first, last + 1))
|
||||
return result
|
||||
|
||||
|
||||
def benign(
|
||||
repo: Path, fixed: str, sink_file: str, extensions: tuple[str, ...], strict: bool = False, vulnerable: str = ""
|
||||
) -> dict[str, object] | None:
|
||||
"""The first acceptable change after the fix (see the module docstring).
|
||||
|
||||
`strict` (a single-language population, v3) tightens the proposal: the change is measured against its first
|
||||
parent, so a merge counts with everything it brings in, and every message it brings in is checked for security
|
||||
words (a branch-sync merge can carry the fix itself); it must change a file of the population's language; and
|
||||
when no commit after the fix qualifies, the history before the vulnerable pin is searched, newest first."""
|
||||
head = git(repo, "rev-parse", "origin/HEAD", check=False).strip() or git(repo, "rev-parse", "HEAD").strip()
|
||||
commits = git(repo, "rev-list", "--first-parent", "--reverse", f"{fixed}..{head}", check=False).split()[:400]
|
||||
span = [f"{fixed}..{head}"]
|
||||
if subprocess.run(["git", "-C", str(repo), "merge-base", "--is-ancestor", fixed, head], check=False).returncode:
|
||||
# A fix on a maintenance branch (population v3: kirby 4.x, roundcube 1.6): the default branch's commits
|
||||
# after the fix date, not its whole history outside the branch.
|
||||
span = [f"--since={git(repo, 'log', '-1', '--format=%cI', fixed).strip()}", head]
|
||||
commits = git(repo, "rev-list", "--first-parent", "--reverse", *span, check=False).split()[:400]
|
||||
after = set(commits)
|
||||
if strict and vulnerable:
|
||||
commits += git(repo, "rev-list", "--first-parent", "--max-count=400", f"{vulnerable}^", check=False).split()
|
||||
for commit in commits:
|
||||
message = git(repo, "log", "-1", "--format=%s%n%b", commit)
|
||||
parent = git(repo, "rev-parse", f"{commit}^", check=False).strip()
|
||||
if not parent:
|
||||
continue
|
||||
if strict:
|
||||
message = git(repo, "log", "--format=%s%n%b", f"{parent}..{commit}")
|
||||
stat = git(repo, "diff", "--numstat", parent, commit).splitlines()
|
||||
else:
|
||||
message = git(repo, "log", "-1", "--format=%s%n%b", commit)
|
||||
stat = git(repo, "show", "--numstat", "--format=", commit).splitlines()
|
||||
if SECURITY_WORDS.search(message):
|
||||
continue
|
||||
stat = git(repo, "show", "--numstat", "--format=", commit).splitlines()
|
||||
files = [line.split("\t") for line in stat if line.count("\t") == 2]
|
||||
if not files or any(f[2] == sink_file for f in files):
|
||||
continue
|
||||
source = [f for f in files if f[2].endswith(extensions)]
|
||||
lines = sum(int(a) + int(b) for a, b, _ in files if a.isdigit() and b.isdigit())
|
||||
if source and 0 < lines <= MAX_BENIGN_LINES:
|
||||
parent = git(repo, "rev-parse", f"{commit}^").strip()
|
||||
return {"base": parent, "tip": commit, "subject": message.splitlines()[0][:120], "files": [f[2] for f in files], "lines": lines}
|
||||
subject = git(repo, "log", "-1", "--format=%s", commit).strip()
|
||||
side = "after the fix" if commit in after else "before the vulnerable pin"
|
||||
return {"base": parent, "tip": commit, "subject": subject[:120], "files": [f[2] for f in files], "lines": lines, "side": side}
|
||||
return None
|
||||
|
||||
|
||||
@@ -112,7 +269,22 @@ def verify(case: dict[str, object], cache: Path) -> dict[str, object]:
|
||||
record["vulnerable_is_ancestor_of_fixed"] = ordered
|
||||
if not ordered:
|
||||
problems.append("the vulnerable pin is not an ancestor of the fixed pin")
|
||||
changed = git(repo, "diff", "--name-only", vulnerable, fixed).splitlines()
|
||||
sites = [str(site) for site in case.get("sites", [])] # type: ignore[union-attr]
|
||||
if case.get("site_files"):
|
||||
# Population v3: files the advisory names without a function are resolved from the fix's diff.
|
||||
record.update(resolve_sites(repo, case, changed))
|
||||
sites = record["sites"] # type: ignore[assignment]
|
||||
if not sites:
|
||||
problems.append("no declared site: the advisory names no function and the fix changes no named file")
|
||||
if case.get("fix_named_by_advisory") is False:
|
||||
# An inferred fix must change the declared sink file and at least one declared site function.
|
||||
basis = record.get("site_basis") or {}
|
||||
named = [s for s in sites if basis.get(s, "advisory-named") == "advisory-named"] # type: ignore[union-attr]
|
||||
touched = fix_changes_sites(repo, case, named)
|
||||
record["inferred_fix_changes_sites"] = touched
|
||||
if not any(touched.values()):
|
||||
problems.append("the inferred fix changes none of the declared site functions: drop the case")
|
||||
if sites:
|
||||
# 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.
|
||||
@@ -139,7 +311,6 @@ def verify(case: dict[str, object], cache: Path) -> dict[str, object]:
|
||||
problems.append(f"none of {functions} appears in {sink_file} at the vulnerable pin")
|
||||
if after is not None:
|
||||
record["sink_sha256_fixed"] = hashlib.sha256(after).hexdigest()
|
||||
changed = git(repo, "diff", "--name-only", vulnerable, fixed).splitlines()
|
||||
record["fix_changed_files"] = len(changed)
|
||||
record["fix_touches_sink_file"] = sink_file in changed
|
||||
fix_site = str(case.get("fix_site", "")).partition("::")[0].strip()
|
||||
@@ -148,7 +319,10 @@ def verify(case: dict[str, object], cache: Path) -> dict[str, object]:
|
||||
problems.append(f"the fix changes neither {sink_file} nor a declared fix_site")
|
||||
languages = [str(case["language"]), *[str(x) for x in case.get("also", [])]] # type: ignore[union-attr]
|
||||
extensions = tuple(e for lang in languages for e in EXTENSIONS.get(lang, ()))
|
||||
record["benign_proposal"] = benign(repo, fixed, sink_file, extensions)
|
||||
strict = bool(case.get("_population_language"))
|
||||
if strict:
|
||||
extensions = EXTENSIONS.get(str(case["_population_language"]), extensions)
|
||||
record["benign_proposal"] = benign(repo, fixed, sink_file, extensions, strict, vulnerable)
|
||||
if record["benign_proposal"] is None and not case.get("benign"):
|
||||
problems.append("no benign change found automatically; choose one by hand")
|
||||
return record
|
||||
@@ -161,7 +335,11 @@ def main() -> int:
|
||||
parser.add_argument("--population", type=Path, default=POPULATION, help="a population-vN.toml in this directory")
|
||||
args = parser.parse_args()
|
||||
population = args.population if args.population.is_absolute() else HERE / args.population.name
|
||||
cases = tomllib.loads(population.read_text())["case"]
|
||||
data = tomllib.loads(population.read_text())
|
||||
cases = data["case"]
|
||||
for case in cases:
|
||||
# A single-language population (v3) gets the strict benign proposal (see `benign`).
|
||||
case["_population_language"] = data.get("language", "")
|
||||
only = {c.strip() for c in args.only.split(",") if c.strip()}
|
||||
records = []
|
||||
for case in cases:
|
||||
|
||||
Reference in New Issue
Block a user