| |
| """Part A (a)-(c): RustSec advisory-db -> candidate pre-fix/post-fix crate-version pairs. |
| |
| Stages (each counted in out/advisory_funnel.json): |
| parsed every crates/*/RUSTSEC-*.md advisory |
| has_patch not informational, has [versions].patched, not "malicious" |
| resolved pre-fix and post-fix versions exist on crates.io (sparse index), pre != post |
| downloaded both .crate tarballs fetched and unpacked |
| small_diff src/**/*.rs diff <= MAX_DIFF_LINES changed lines and >=1 changed .rs file |
| diff_safe changed lines contain no unsafe/async/io/fs/net/thread/extern/ffi tokens |
| strict_safe (subset of diff_safe) changed files contain no `unsafe` at all |
| The Charon/Aeneas translation stage is run by translate_items.py afterwards. |
| |
| Usage: select_advisories.py [--db DIR] [--data DIR] [--out FILE] [--limit N] |
| No state is written under /home or /tmp (urllib + a data dir under /apps). |
| """ |
| import argparse, json, os, re, subprocess, sys, tarfile, tomllib, urllib.request, io |
| from concurrent.futures import ThreadPoolExecutor |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| import semverlite as sv |
|
|
| MAX_DIFF_LINES = 300 |
| UA = {"User-Agent": "align-bugfind-bench (tgptsupport@trustt.com)"} |
| BAD = re.compile(r"\bunsafe\b|\basync\b|\.await\b|std::(io|fs|net|thread|process|env)\b|\bextern\b|\bffi\b|\bFile\b|\bTcp|\bUdp|libc::|\bmpsc\b|\bMutex\b|\bArc\b") |
| CLASS = [("overflow", r"overflow|underflow|wrap"), ("panic", r"\bpanic|unwrap|abort|assert|index out|out[- ]of[- ]bounds|slice"), |
| ("dos_loop_or_recursion", r"infinite loop|stack overflow|recursion|exhaust|unbounded|quadratic|hang"), |
| ("logic", r"incorrect|wrong|bypass|validation|parse|parsing|decod")] |
|
|
|
|
| def http(url): |
| req = urllib.request.Request(url, headers=UA) |
| with urllib.request.urlopen(req, timeout=60) as r: |
| return r.read() |
|
|
|
|
| def index_path(name): |
| n = name.lower() |
| if len(n) == 1: return f"1/{n}" |
| if len(n) == 2: return f"2/{n}" |
| if len(n) == 3: return f"3/{n[0]}/{n}" |
| return f"{n[:2]}/{n[2:4]}/{n}" |
|
|
|
|
| def parse_advisories(db): |
| out = [] |
| for p in sorted(os.scandir(os.path.join(db, "crates")), key=lambda e: e.name): |
| for f in sorted(os.listdir(p.path)): |
| if not f.endswith(".md"): continue |
| txt = open(os.path.join(p.path, f), encoding="utf8").read() |
| m = re.match(r"```toml\n(.*?)\n```\n(.*)", txt, re.S) |
| if not m: continue |
| try: |
| t = tomllib.loads(m.group(1)) |
| except Exception: |
| continue |
| a = t.get("advisory", {}) |
| out.append(dict(id=a.get("id"), package=a.get("package"), date=str(a.get("date")), url=a.get("url"), |
| categories=a.get("categories", []), keywords=a.get("keywords", []), |
| informational=a.get("informational"), withdrawn=a.get("withdrawn"), |
| patched=t.get("versions", {}).get("patched", []), |
| unaffected=t.get("versions", {}).get("unaffected", []), |
| functions=sorted(t.get("affected", {}).get("functions", {}).keys()), |
| title=m.group(2).strip().split("\n")[0].lstrip("# "), body=m.group(2)[:1500])) |
| return out |
|
|
|
|
| def resolve(adv, cache): |
| name = adv["package"] |
| try: |
| raw = http("https://index.crates.io/" + index_path(name)).decode() |
| except Exception as e: |
| return None, f"index:{e}" |
| vers = [] |
| for line in raw.splitlines(): |
| j = json.loads(line) |
| if j.get("yanked") or sv.is_pre(j["vers"]): continue |
| vers.append(j["vers"]) |
| vers.sort(key=sv.key) |
| patched = [v for v in vers if sv.matches_any(v, adv["patched"])] |
| if not patched: return None, "no_published_patched_version" |
| post = patched[0] |
| cands = [v for v in vers if sv.key(v) < sv.key(post) and not sv.matches_any(v, adv["patched"]) and not sv.matches_any(v, adv["unaffected"])] |
| if not cands: return None, "no_published_vulnerable_version" |
| return (cands[-1], post), None |
|
|
|
|
| def fetch_crate(name, ver, data): |
| d = os.path.join(data, "crates", f"{name}-{ver}") |
| if os.path.isdir(d): return d |
| blob = http(f"https://static.crates.io/crates/{name}/{name}-{ver}.crate") |
| os.makedirs(os.path.join(data, "crates"), exist_ok=True) |
| tmp = d + ".part" |
| with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tf: |
| tf.extractall(tmp, filter="data") |
| os.rename(os.path.join(tmp, f"{name}-{ver}"), d) |
| os.rmdir(tmp) |
| return d |
|
|
|
|
| def rs_files(root): |
| r = {} |
| for dp, _, fs in os.walk(os.path.join(root, "src")): |
| for f in fs: |
| if f.endswith(".rs"): |
| p = os.path.join(dp, f) |
| r[os.path.relpath(p, root)] = open(p, encoding="utf8", errors="replace").read() |
| return r |
|
|
|
|
| def analyse_diff(pre, post): |
| import difflib |
| a, b = rs_files(pre), rs_files(post) |
| changed, lines, bad_lines = [], 0, [] |
| for f in sorted(set(a) | set(b)): |
| if a.get(f) == b.get(f): continue |
| changed.append(f) |
| for l in difflib.unified_diff(a.get(f, "").splitlines(), b.get(f, "").splitlines(), lineterm="", n=0): |
| if l.startswith(("+++", "---", "@@")): continue |
| lines += 1 |
| if BAD.search(l): bad_lines.append(l[:120]) |
| file_unsafe = sum(len(re.findall(r"\bunsafe\b", (a.get(f, "") + b.get(f, "")))) for f in changed) |
| return dict(changed_files=changed, changed_lines=lines, bad_line_count=len(bad_lines), bad_line_samples=bad_lines[:3], unsafe_in_changed_files=file_unsafe) |
|
|
|
|
| def classify(adv): |
| t = (adv["title"] + " " + " ".join(adv["keywords"]) + " " + adv["body"]).lower() |
| return [c for c, rx in CLASS if re.search(rx, t)] or ["other"] |
|
|
|
|
| def process(adv, data): |
| rec = dict(adv_id=adv["id"], crate=adv["package"], date=adv["date"], url=adv["url"], title=adv["title"], categories=adv["categories"], |
| affected_functions=adv["functions"], bug_class=classify(adv)) |
| r, why = resolve(adv, None) |
| if r is None: |
| rec["stage"], rec["reject"] = "has_patch", why; return rec |
| rec["pre_version"], rec["post_version"] = r |
| try: |
| pre = fetch_crate(adv["package"], r[0], data); post = fetch_crate(adv["package"], r[1], data) |
| except Exception as e: |
| rec["stage"], rec["reject"] = "resolved", f"download:{e}"; return rec |
| rec["pre_dir"], rec["post_dir"] = pre, post |
| rec.update(analyse_diff(pre, post)) |
| if not rec["changed_files"]: rec["stage"], rec["reject"] = "downloaded", "no_rs_diff"; return rec |
| if rec["changed_lines"] > MAX_DIFF_LINES: rec["stage"], rec["reject"] = "downloaded", "diff_too_large"; return rec |
| rec["stage"] = "small_diff" |
| if rec["bad_line_count"]: rec["reject"] = "diff_touches_unsafe_io_async"; return rec |
| rec["stage"] = "diff_safe" |
| if rec["unsafe_in_changed_files"] == 0: rec["stage"] = "strict_safe" |
| return rec |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--db", default="/apps/development/rlhf/bugfind_data/advisory-db") |
| ap.add_argument("--data", default="/apps/development/rlhf/bugfind_data") |
| ap.add_argument("--out", default=None) |
| ap.add_argument("--limit", type=int, default=0) |
| a = ap.parse_args() |
| if not a.data.startswith("/apps/"): sys.exit("data dir must be under /apps") |
| out = a.out or os.path.join(a.data, "out", "advisory_candidates.jsonl") |
| os.makedirs(os.path.dirname(out), exist_ok=True) |
| advs = parse_advisories(a.db) |
| funnel = {"parsed": len(advs)} |
| keep = [x for x in advs if x["patched"] and not x["informational"] and not x["withdrawn"] and "malicious" not in x["categories"]] |
| funnel["has_patch"] = len(keep) |
| if a.limit: keep = keep[: a.limit] |
| with ThreadPoolExecutor(4) as ex: |
| recs = list(ex.map(lambda x: process(x, a.data), keep)) |
| funnel["resolved"] = sum(1 for r in recs if "pre_version" in r) |
| funnel["downloaded"] = sum(1 for r in recs if "pre_dir" in r) |
| funnel["small_diff"] = sum(1 for r in recs if r["stage"] in ("small_diff", "diff_safe", "strict_safe")) |
| funnel["diff_safe"] = sum(1 for r in recs if r["stage"] in ("diff_safe", "strict_safe")) |
| funnel["strict_safe"] = sum(1 for r in recs if r["stage"] == "strict_safe") |
| funnel["reject_reasons"] = {} |
| for r in recs: |
| if r.get("reject"): funnel["reject_reasons"][r["reject"].split(":")[0]] = funnel["reject_reasons"].get(r["reject"].split(":")[0], 0) + 1 |
| with open(out, "w") as f: |
| for r in recs: f.write(json.dumps(r) + "\n") |
| json.dump(funnel, open(os.path.join(os.path.dirname(out), "advisory_funnel.json"), "w"), indent=1) |
| print(json.dumps(funnel)) |
|
|
| if __name__ == "__main__": |
| main() |
|
|
