#!/usr/bin/env python3 -I """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()