File size: 13,062 Bytes
8a2e047 | 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 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 | """Audit corpus coverage and freeze reproducible, position-balanced evaluations."""
import hashlib
import json
import random
import re
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parent
OUT = ROOT / "benchmarks"
SEED = 20261109
SECEVAL_REVISION = "205dab7b0888a06f4b53ca7d9c7093e1326683e1"
CVE = re.compile(r"CVE-\d{4}-\d{4,}", re.I)
TOPICS = {
"iOS / Apple platform": r"\b(?:ios|ipados|xnu|iokit|webkit|darwin|iphone)\b",
"Kernel / operating systems": r"\b(?:kernel|syscall|sandbox|system call|privilege escalation|linux|windows|android)\b",
"Memory safety / binary analysis": r"\b(?:buffer overflow|out.of.bounds|use.after.free|heap|stack|rop|disassembl\w*|fuzz\w*)\b",
"Web / API security": r"\b(?:sql injection|xss|ssrf|csrf|http|web|api|cookies?)\b",
"Network security": r"\b(?:network|tcp|udp|dns|firewall|packet|wi.fi|wireless|vpn)\b",
"Identity / authentication": r"\b(?:authentication|oauth|saml|kerberos|mfa|active directory|credentials?)\b",
"Cryptography": r"\b(?:cryptograph\w*|encryption|cipher|tls|ssl|hash|certificate)\b",
"Cloud / containers": r"\b(?:aws|azure|cloud|docker|kubernetes|container)\b",
"Malware / incident analysis": r"\b(?:malware|ransomware|forensic\w*|incident|yara|trojan)\b",
}
def read_rows(split):
return [json.loads(line) for line in (ROOT / "data" / f"{split}.jsonl").read_text(encoding="utf-8").splitlines()]
def normalized(text):
return " ".join(re.findall(r"\w+", text.casefold()))
def shingles(text):
words = normalized(text).split()
return {tuple(words[i:i+4]) for i in range(max(0, len(words)-3))}
def parse_local(row):
stem, options = row["prompt"].split("\n\nA) ", 1)
options = "A) " + options.rsplit("\n\nChoose the best answer and explain it.", 1)[0]
matches = re.findall(r"(?ms)^([A-D])\) (.*?)(?=^[A-D]\) |\Z)", options)
assert [m[0] for m in matches] == list("ABCD"), row["group"]
match = re.match(r"^([A-D])\)", row["response"])
assert match, row["group"]
return stem, [m[1].strip() for m in matches], "ABCD".index(match.group(1))
def scoring_prefix(tokenizer, question, choices):
prompt = question + "\n\n" + "\n".join(f"{chr(65+i)}) {choice}" for i, choice in enumerate(choices))
prompt += "\n\nSelect the best answer. Reply with only its letter."
prefix = tokenizer.apply_chat_template([{"role":"user", "content":prompt}], tokenize=False, add_generation_prompt=True)
assert prefix.endswith("<think>\n")
return prefix + "\n</think>\n\n"
def coverage(train, evaluation):
result = {"created_utc": datetime.now(timezone.utc).isoformat(), "definitions": {
"ios_versions": "Release containing the fix, as named by Apple; not an affected-version range.",
"topics": "Non-exclusive keyword matches in prompt plus answer, not expert topic labels or measured model competence.",
"unique_cves": "Distinct CVE identifiers mentioned in retained records; not a fraction of all publicly known CVEs."}}
for name, rows in [("train", train), ("eval", evaluation)]:
versions, components, vendors, products = Counter(), Counter(), Counter(), Counter()
for row in rows:
if row["kind"] == "apple_advisory":
versions[re.search(r"Advisory: iOS ([\d.]+)", row["prompt"]).group(1)] += 1
excerpt = row["prompt"].split("\n\n")[1].splitlines()
component = excerpt[0] if len(excerpt) > 1 and excerpt[1].startswith("Available for:") else "Unparsed component"
components[component] += 1
elif row["kind"] == "cisa_kev":
record = json.loads(row["prompt"].split("\n\n")[1])
vendors[record["vendorProject"]] += 1
products[record["vendorProject"] + " / " + record["product"]] += 1
labels = Counter(row["response"][0] for row in rows if row["kind"] == "pentesting_explanation")
unique_cves = {cve for row in rows for cve in row["cves"]}
result[name] = dict(records=len(rows), by_source=dict(Counter(r["kind"] for r in rows)),
unique_cves=len(unique_cves), cve_years=dict(Counter(c.split("-")[1] for c in unique_cves)),
ios_versions=dict(versions), ios27_examples=sum(n for v,n in versions.items() if v.split(".")[0] == "27"),
apple_components=dict(components.most_common()), cisa_vendors=dict(vendors.most_common()),
cisa_distinct_vendors=len(vendors), cisa_distinct_products=len(products),
mcq_original_answer_labels=dict(labels),
keyword_topics={topic:sum(bool(re.search(pattern,r["prompt"]+"\n"+r["response"],re.I)) for r in rows)
for topic,pattern in TOPICS.items()})
result["split_overlap"] = {
"groups": len({r["group"] for r in train} & {r["group"] for r in evaluation}),
"cves": len({c for r in train for c in r["cves"]} & {c for r in evaluation for c in r["cves"]})}
result["corpus_sha256"] = {s:hashlib.sha256((ROOT/"data"/f"{s}.jsonl").read_bytes()).hexdigest() for s in ["train","eval"]}
result["gaps"] = [
"No iOS 27 advisory examples are in the current held-out split; iOS 27 generalization remains unmeasured.",
"Apple examples primarily teach advisory interpretation, not kernel/WebKit source-code auditing.",
"Original MCQ answer positions are heavily skewed toward A; the separate benchmark reorders eligible choices to balance answer letters.",
"Exact group/CVE and lexical overlap checks cannot rule out semantic overlap or base-model pretraining contamination."]
return result
def prepare():
from transformers import AutoTokenizer
OUT.mkdir(exist_ok=True)
train, evaluation = read_rows("train"), read_rows("eval")
report = coverage(train,evaluation)
(OUT/"coverage.json").write_text(json.dumps(report,indent=2),encoding="utf-8")
tokenizer = AutoTokenizer.from_pretrained(ROOT/"base", local_files_only=True)
train_questions = [parse_local(r)[0] if r["kind"] == "pentesting_explanation" else r["prompt"] for r in train]
exact = {normalized(q) for q in train_questions}
train_grams = [shingles(q) for q in train_questions]
index = defaultdict(set)
for i, grams in enumerate(train_grams):
for gram in grams:
index[gram].add(i)
train_cves = {c for r in train for c in r["cves"]}
excluded = Counter()
pools = defaultdict(list)
def add(suite, case_id, question, choices, answer, topics, source):
if len(choices) != 4 or len(set(choices)) != 4 or not 0 <= answer < 4:
excluded[suite+":invalid_choices"] += 1
return
# Reordering must not change the meaning of positional references.
position = r"\b(?:above|below|preceding|following options|first two|last two)\b|\b(?:option|choice|statement)s?\s+[A-D]\b|\b[A-D]\s+(?:and|or)\s+[A-D]\b"
if re.search(position, question + "\n" + "\n".join(choices), re.I):
excluded[suite+":position_dependent_options"] += 1
return
if normalized(question) in exact or train_cves.intersection(c.upper() for c in CVE.findall(question+" "+" ".join(choices))):
excluded[suite+":training_overlap"] += 1
return
grams = shingles(question)
candidates = set().union(*(index.get(g,set()) for g in grams)) if grams else set()
if any(len(grams & train_grams[i])/max(1,len(grams | train_grams[i])) >= 0.8 for i in candidates):
excluded[suite+":lexical_near_duplicate"] += 1
return
if len(tokenizer(scoring_prefix(tokenizer,question,choices),add_special_tokens=False)["input_ids"]) > 768:
excluded[suite+":overlength"] += 1
return
pools[suite].append(dict(id=suite+":"+str(case_id), question=question, choices=choices,
answer_index=answer,topics=topics,source=source))
for row in evaluation:
if row["kind"] == "pentesting_explanation":
stem, choices, answer = parse_local(row)
add("heldout_pentesting",row["group"],stem,choices,answer,["Pentesting"],row["source"])
seceval_file = ROOT/"data"/"benchmark-sources"/"SecEval"/"questions.json"
for row in json.loads(seceval_file.read_text(encoding="utf-8")):
if row["answer"].strip() not in list("ABCD"):
excluded["seceval_200:multiple_or_invalid_answer"] += 1
continue
choices = [re.sub(r"^[A-D]\s*[:.)]\s*", "", choice).strip() for choice in row["choices"]]
add("seceval_200",row["id"],row["question"],choices,"ABCD".index(row["answer"].strip()),
row.get("topics") or ["Unspecified"],"https://huggingface.co/datasets/XuanwuAI/SecEval/tree/"+SECEVAL_REVISION)
selected = []
eligible = {name:len(pool) for name,pool in pools.items()}
for name, pool in sorted(pools.items()):
rng = random.Random(SEED)
rng.shuffle(pool)
if name == "seceval_200":
# Round-robin by primary topic to sample across available domains.
buckets = defaultdict(list)
for row in pool:
buckets[row["topics"][0]].append(row)
ordered = []
while any(buckets.values()):
for topic in sorted(buckets):
if buckets[topic]:
ordered.append(buckets[topic].pop())
pool = ordered
limit = min(200 if name == "seceval_200" else len(pool), len(pool))//4*4
if name != "seceval_200" and len(pool) > limit:
excluded[name+":balance_rounding"] += len(pool)-limit
targets = list(range(4))*(limit//4)
rng.shuffle(targets)
for row, target in zip(pool[:limit],targets):
original_answer = row["answer_index"]
original_choices = row["choices"]
distractors = [i for i in range(4) if i != original_answer]
rng.shuffle(distractors)
distractors.insert(target,original_answer)
row.update(choices=[original_choices[i] for i in distractors],answer_index=target,
answer_letter=chr(65+target),original_choice_indices=distractors,
question_sha256=hashlib.sha256(row["question"].encode("utf-8")).hexdigest())
prefix = scoring_prefix(tokenizer,row["question"],row["choices"])
ids = tokenizer(prefix,add_special_tokens=False)["input_ids"]
for letter in "ABCD":
full = tokenizer(prefix+letter,add_special_tokens=False)["input_ids"]
assert full[:-1] == ids and len(full) == len(ids)+1, "Letter ranking requires stable single-token continuations"
row["prompt_tokens"] = len(ids)
selected.append(row)
suite = dict(protocol="balanced-letter-ranking-v1",seed=SEED,cases=selected,
corpus_sha256=report["corpus_sha256"],seceval_revision=SECEVAL_REVISION,
seceval_file_sha256=hashlib.sha256(seceval_file.read_bytes()).hexdigest(),
source_attribution="SecEval: XuanwuAI, CC BY-NC-SA; Pentesting Explanations: Christopher Dickinson, Apache-2.0. Evaluation only.")
suite_bytes = json.dumps(suite,indent=2,ensure_ascii=False).encode("utf-8")
(OUT/"suite.json").write_bytes(suite_bytes)
plan = dict(protocol=suite["protocol"],seed=SEED,suite_sha256=hashlib.sha256(suite_bytes).hexdigest(),
corpus_sha256=report["corpus_sha256"],seceval_revision=SECEVAL_REVISION,
seceval_file_sha256=suite["seceval_file_sha256"],
status="queued_after_training",total_questions=len(selected),excluded=dict(excluded),eligible_pool_sizes=eligible,
method="Argmax of next-token A/B/C/D logits after a closed reasoning prefix; no generated reasoning. Paired base and adapter in the same NF4 runtime.",
comparison="Adapted subsets, not official SecEval leaderboard scores or generated-answer pass@1.",
suites={name:dict(questions=sum(r["id"].startswith(name+":") for r in selected),
answer_labels=dict(Counter(r["answer_letter"] for r in selected if r["id"].startswith(name+":"))),
topic_counts=dict(Counter(t for r in selected if r["id"].startswith(name+":") for t in r["topics"]))) for name in pools},
cases=[{k:r[k] for k in ["id","answer_letter","topics","source","prompt_tokens",
"original_choice_indices","question_sha256"]} for r in selected])
assert all(len(set(s["answer_labels"].values())) == 1 for s in plan["suites"].values())
assert len(selected) >= 300
(OUT/"plan.json").write_text(json.dumps(plan,indent=2),encoding="utf-8")
print(json.dumps({"coverage":{s:{k:report[s][k] for k in ["records","unique_cves","ios27_examples","cisa_distinct_vendors","cisa_distinct_products"]} for s in ["train","eval"]},
"benchmark_questions":plan["total_questions"],"suites":plan["suites"],"excluded":plan["excluded"]},indent=2))
if __name__ == "__main__":
prepare()
|