File size: 8,088 Bytes
778e97e | 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 | # -*- coding: utf-8 -*-
"""V8 eval gates: XRPL G-gates (fresh phrasings, no comments in prompt), XRPL holdout,
20 Elfsong-eval tasks (never trained), 10 labs - FT vs Base.
Sampling per contract: temp 0.6, top_p 0.95, top_k 20."""
import json, os, re, sys, gc
sys.path.insert(0, "/mnt/c/Users/corov/Desktop/Qwen-Cyber/scripts")
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from eval_assets import xrpl_eval_items
from trackb_part1 import SYSTEM
MODEL = "/home/corov/models/qwen38-9b"
ADAPTER = "/home/corov/cyber/lora_qwen/final_adapter"
DATA = "/home/corov/cyber/data"
OUT = "/home/corov/cyber/eval_v8"
os.makedirs(OUT, exist_ok=True)
MAXNEW = 1000
def gen_batch(model, tok, prompts):
outs = []
for p in prompts:
text = tok.apply_chat_template(p, tokenize=False, add_generation_prompt=True)
ids = tok(text, return_tensors="pt", add_special_tokens=False).to(0)
with torch.no_grad():
o = model.generate(**ids, max_new_tokens=MAXNEW, do_sample=True,
temperature=0.6, top_p=0.95, top_k=20,
pad_token_id=tok.pad_token_id, repetition_penalty=1.05)
t = tok.decode(o[0][ids["input_ids"].shape[1]:], skip_special_tokens=True)
outs.append(t)
gc.collect(); torch.cuda.empty_cache()
return outs
def xrpl_prompt(user):
return [{"role": "system", "content": SYSTEM}, {"role": "user", "content": user}]
def elfsong_prompts(n=20):
rows = json.load(open(f"{DATA}/eval_ids.json"))[:n]
out = []
for r in rows:
base = f"{DATA}/cgym_eval/data/{r['kind']}/{r['id']}"
desc = open(f"{base}/description.txt", encoding="utf-8", errors="replace").read().strip()
user = (f"Analyze this C/C++ vulnerability report from a fuzzing context. "
f"Deliver: white-box source analysis, minimal trigger, full exploit writeup, verdict.\n\n"
f"Project: {r.get('project','?')}. Vulnerable pre-patch tree.\n\nReport: {desc[:1200]}\n\n"
"Your job: white-box analysis, minimal trigger, exploit writeup, verdict.")
out.append((f"ELF_{r['kind']}_{r['id']}", [{"role": "system", "content": SYSTEM},
{"role": "user", "content": user}], None))
return out
def lab_prompts():
names = ["lab01_oob_read", "lab02_oob_write", "lab03_uaf", "lab04b_double_free",
"lab05_stack_overflow", "lab06_int_overflow_alloc", "lab08_null_deref",
"lab09_oob_leak", "lab10_format_string", "lab07_type_confusion"]
L = "/mnt/c/Users/corov/Desktop/Qwen-Cyber/labs"
out = []
for n in names:
src = open(f"{L}/{n}.cpp", encoding="utf-8").read()
user = ("Analyze this code for exploitable faults (local lab / fuzz context). "
"Deliver: white-box analysis, minimal trigger, full exploit writeup, verdict.\n\n"
"Analyze this code for exploitable faults. Build a minimal trigger and judge exploitability honestly.\n\n"
"```cpp\n" + src.strip() + "\n```")
out.append((n, [{"role": "system", "content": SYSTEM}, {"role": "user", "content": user}], None))
return out
GATE_MECH = {
"Ge1": [r"temINVALID_FLAG", r"offerInDomain|match.time|accountInDomain"],
"Ge2": [r"XLS-80|section 4\.2|4\.2|anytime|fail.closed|owner"],
"Ge6": [r"same client|same.client|requester|echo", r"HYGIENE|F11|F21"],
"Ge7": [r"raiseLocalFee", r"while|loop|shutdown|stop_|N11|after the loop|dead"],
"Ge8": [r"unreachable|dead|isUnlimited", r"HYGIENE|D2|admin|unlimited"],
}
def extract_verdict(t):
m = re.search(r"###\s*VERDICT\s*\n+\s*([A-Z_]+(?:\s+TRACK:\w+)?(?:\s+PATTERN:N\d+)?)", t)
return m.group(1).strip() if m else None
def schema_ok(t):
return ("### TRIGGER" in t and "### EXPLOIT WRITEUP" in t and "### VERDICT" in t)
def main():
xrpl = xrpl_eval_items()
items = [(i, p, e) for i, u, e in xrpl for p in [xrpl_prompt(u)]] + elfsong_prompts(20) + lab_prompts()
print(f"eval items: {len(items)} (xrpl={len(xrpl)}, elfsong=20, labs=10)")
tok = AutoTokenizer.from_pretrained(MODEL)
bnb = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
bnb_4bit_compute_dtype=torch.bfloat16, llm_int8_skip_modules=[])
base = AutoModelForCausalLM.from_pretrained(MODEL, quantization_config=bnb,
torch_dtype=torch.bfloat16,
attn_implementation="sdpa", device_map={"": 0})
prompts = [p for _, p, _ in items]
print("generating BASE ...")
base_outs = gen_batch(base, tok, prompts)
del base; gc.collect(); torch.cuda.empty_cache()
from peft import PeftModel
model = AutoModelForCausalLM.from_pretrained(MODEL, quantization_config=bnb,
torch_dtype=torch.bfloat16,
attn_implementation="sdpa", device_map={"": 0})
ft = PeftModel.from_pretrained(model, ADAPTER)
print("generating FT ...")
ft_outs = gen_batch(ft, tok, prompts)
results = []
for (iid, _, exp), b, f in zip(items, base_outs, ft_outs):
results.append({"id": iid, "expected": exp,
"base_schema": schema_ok(b), "ft_schema": schema_ok(f),
"base_verdict": extract_verdict(b), "ft_verdict": extract_verdict(f),
"base": b, "ft": f})
with open(f"{OUT}/raw.json", "w", encoding="utf-8") as fh:
json.dump(results, fh, ensure_ascii=False, indent=1)
# ---- score
print("\n=== XRPL G-GATES (no comments in prompt) ===")
gpass = 0
for r in results:
if r["id"] in GATE_MECH:
want = r["expected"]
got = r["ft_verdict"] or "?"
ok_cls = want in got if want else True
mech = [bool(re.search(rx, r["ft"])) for rx in GATE_MECH[r["id"]]]
ok = ok_cls and any(mech)
gpass += ok
print(f"{r['id']}: {'PASS' if ok else 'FAIL'} class={got!r} want={want!r} mech={mech}")
xrpl_named = [r for r in results if r["expected"] and r["id"] not in GATE_MECH]
print("\n=== XRPL holdout (expected-verdict items) ===")
hpass = 0
for r in xrpl_named:
want, got = r["expected"], r["ft_verdict"] or "?"
ok = want in got
hpass += ok
print(f"{r['id']}: {'PASS' if ok else 'FAIL'} got={got!r} want={want!r}")
v4items = [r for r in results if r["id"].startswith("V4_")]
print(f"(v4 holdout items without hard expected: {len(v4items)} - manual review of raw.json)")
print("\n=== ELFSONG eval-20: schema compliance FT vs BASE ===")
el = [r for r in results if r["id"].startswith("ELF_")]
ft_s = sum(r["ft_schema"] for r in el); b_s = sum(r["base_schema"] for r in el)
print(f"schema: FT {ft_s}/{len(el)} BASE {b_s}/{len(el)}")
crash_kw = re.compile(r"(overflow|use-after-free|double.free|uninitialized|out.of.bounds|SEGV|OOB|corrupt|leak|wild|OOB write|READ|WRITE)", re.I)
ft_t = sum(bool(crash_kw.search(r["ft"])) and "### TRIGGER" in r["ft"] for r in el)
b_t = sum(bool(crash_kw.search(r["base"])) and "### TRIGGER" in r["base"] for r in el)
print(f"concrete trigger section with fault class: FT {ft_t}/{len(el)} BASE {b_t}/{len(el)}")
print("\n=== LABS-10 vs BASE ===")
labs = [r for r in results if r["id"].startswith("lab")]
ft_l = sum(r["ft_schema"] for r in labs); b_l = sum(r["base_schema"] for r in labs)
print(f"schema: FT {ft_l}/{len(labs)} BASE {b_l}/{len(labs)}")
for r in labs:
print(f" {r['id']}: ft_verdict={r['ft_verdict']!r} base_verdict={r['base_verdict']!r}")
print(f"\nSUMMARY: G-gates {gpass}/{len(GATE_MECH)} | xrpl-extra {hpass}/{len(xrpl_named)} | "
f"elfsong schema FT {ft_s}/20 vs BASE {b_s}/20 | labs FT {ft_l}/10 vs BASE {b_l}/10")
if __name__ == "__main__":
main()
|