scripts/eval_v8.py

# -*- 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()