File size: 8,769 Bytes
4951e76 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | #!/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()
|