06 · Use case 3: the code vulnerability hunter¶

The problem: a security review has to read every function, and nearly all of them are fine. An LLM reading a whole repo is slow and expensive, and it gets lazy on long inputs.

The fast/slow split is map-reduce:

 repo -> (code) split into functions with `ast` -> MAP: Jev classifies EVERY chunk (CWE class + severity)
                                                        |
                                            REDUCE: the LLM explains + patches ONLY the flagged chunks

The target is data/vulnerable_app/, a toy app with 9 planted bugs across 7 classes (SQL injection, command injection, path traversal, hardcoded secret, unsafe deserialization, weak hashing, XSS) and a safe look-alike for most of them. It is never executed or imported, only parsed. The file tools are the sandboxed read-only tools from the original repo's notebook 04.

In [1]:

# --- Provider config: ONE OpenRouter key drives both brains ------------------------------
#   slow brain (LLM) -> OpenAI SDK   -> https://openrouter.ai/api/v1  (chat completions + tools)
#   fast brain (Jev) -> TypeSafe SDK -> https://openrouter.ai/api     (POST /v1/systemone)
import os, json, time

# Load settings from a .env file if present (falls back to existing env vars).
try:
    from dotenv import load_dotenv, find_dotenv
    load_dotenv(find_dotenv(usecwd=True))
except Exception:
    if os.path.exists(".env"):
        for _line in open(".env"):
            _line = _line.strip()
            if _line and not _line.startswith("#") and "=" in _line:
                _k, _v = _line.split("=", 1)
                os.environ.setdefault(_k.strip(), _v.strip())

BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1")
API_KEY  = os.environ.get("OPENAI_API_KEY", "set-me")
MODEL    = os.environ.get("MODEL", "openai/gpt-6-luna")          # slow brain: any tool-capable chat model

JEV_BACKEND  = os.environ.get("JEV_BACKEND", "typesafe").strip().lower()  # typesafe | adapter
JEV_MODEL    = os.environ.get("JEV_MODEL", "~typesafe/jev-latest")
JEV_BASE_URL = os.environ.get("TYPESAFE_BASE_URL", "https://openrouter.ai/api")
JEV_API_KEY  = os.environ.get("TYPESAFE_API_KEY") or API_KEY          # one key for both brains

# Behind a TLS-intercepting firewall/proxy, set VERIFY_SSL=false in .env (trusted networks only).
import httpx, httpx2
VERIFY_SSL = os.environ.get("VERIFY_SSL", "true").strip().lower() not in ("false", "0", "no")
if not VERIFY_SSL:
    import warnings
    warnings.filterwarnings("ignore")
    print("\u26a0\ufe0f  SSL verification DISABLED (VERIFY_SSL=false) \u2014 use only on a trusted network")

from openai import OpenAI
client = OpenAI(base_url=BASE_URL, api_key=API_KEY, http_client=httpx.Client(verify=VERIFY_SSL))

if JEV_BACKEND == "adapter":
    # No Jev access yet? Same System One API, answered by your LLM (slower, pricier, but it runs).
    from system_one_adapter import SystemOneAdapterClient, Noul, Choice, Score
    from system_one_adapter.providers.openai import OpenAIProvider
    jev = SystemOneAdapterClient(
        structured_outputs=True, llm_answer_mode="probabilities", normalize_probabilities=True,
        n_retry_malformed_structure=2,
        model=OpenAIProvider(MODEL, base_url=BASE_URL, api_key=API_KEY, api="chat_completions"))
else:
    from typesafe_sdk import TypeSafeClient, Noul, Choice, Score
    jev = TypeSafeClient(api_key=JEV_API_KEY, base_url=JEV_BASE_URL, model=JEV_MODEL,
                         http_client=httpx2.Client(verify=VERIFY_SSL))

print(f"slow brain (LLM): {MODEL}  @ {BASE_URL}")
print(f"fast brain (Jev): {JEV_MODEL if JEV_BACKEND != 'adapter' else MODEL + ' via adapter'}  @ {JEV_BASE_URL if JEV_BACKEND != 'adapter' else BASE_URL}")

# --- Provider config: ONE OpenRouter key drives both brains ------------------------------ # slow brain (LLM) -> OpenAI SDK -> https://openrouter.ai/api/v1 (chat completions + tools) # fast brain (Jev) -> TypeSafe SDK -> https://openrouter.ai/api (POST /v1/systemone) import os, json, time # Load settings from a .env file if present (falls back to existing env vars). try: from dotenv import load_dotenv, find_dotenv load_dotenv(find_dotenv(usecwd=True)) except Exception: if os.path.exists(".env"): for _line in open(".env"): _line = _line.strip() if _line and not _line.startswith("#") and "=" in _line: _k, _v = _line.split("=", 1) os.environ.setdefault(_k.strip(), _v.strip()) BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1") API_KEY = os.environ.get("OPENAI_API_KEY", "set-me") MODEL = os.environ.get("MODEL", "openai/gpt-6-luna") # slow brain: any tool-capable chat model JEV_BACKEND = os.environ.get("JEV_BACKEND", "typesafe").strip().lower() # typesafe | adapter JEV_MODEL = os.environ.get("JEV_MODEL", "~typesafe/jev-latest") JEV_BASE_URL = os.environ.get("TYPESAFE_BASE_URL", "https://openrouter.ai/api") JEV_API_KEY = os.environ.get("TYPESAFE_API_KEY") or API_KEY # one key for both brains # Behind a TLS-intercepting firewall/proxy, set VERIFY_SSL=false in .env (trusted networks only). import httpx, httpx2 VERIFY_SSL = os.environ.get("VERIFY_SSL", "true").strip().lower() not in ("false", "0", "no") if not VERIFY_SSL: import warnings warnings.filterwarnings("ignore") print("\u26a0\ufe0f SSL verification DISABLED (VERIFY_SSL=false) \u2014 use only on a trusted network") from openai import OpenAI client = OpenAI(base_url=BASE_URL, api_key=API_KEY, http_client=httpx.Client(verify=VERIFY_SSL)) if JEV_BACKEND == "adapter": # No Jev access yet? Same System One API, answered by your LLM (slower, pricier, but it runs). from system_one_adapter import SystemOneAdapterClient, Noul, Choice, Score from system_one_adapter.providers.openai import OpenAIProvider jev = SystemOneAdapterClient( structured_outputs=True, llm_answer_mode="probabilities", normalize_probabilities=True, n_retry_malformed_structure=2, model=OpenAIProvider(MODEL, base_url=BASE_URL, api_key=API_KEY, api="chat_completions")) else: from typesafe_sdk import TypeSafeClient, Noul, Choice, Score jev = TypeSafeClient(api_key=JEV_API_KEY, base_url=JEV_BASE_URL, model=JEV_MODEL, http_client=httpx2.Client(verify=VERIFY_SSL)) print(f"slow brain (LLM): {MODEL} @ {BASE_URL}") print(f"fast brain (Jev): {JEV_MODEL if JEV_BACKEND != 'adapter' else MODEL + ' via adapter'} @ {JEV_BASE_URL if JEV_BACKEND != 'adapter' else BASE_URL}")

slow brain (LLM): openai/gpt-6-luna  @ https://openrouter.ai/api/v1
fast brain (Jev): ~typesafe/jev-latest  @ https://openrouter.ai/api

In [2]:

# --- Helpers used in every notebook: ask the fast brain, ask the slow brain, track spend ---
class Spend:
    """Running tally of what OpenRouter charged, per brain (it reports exact USD per call)."""
    def __init__(self):
        self.usd, self.calls = {"jev": 0.0, "llm": 0.0}, {"jev": 0, "llm": 0}
    def add(self, brain, resp):
        self.calls[brain] += 1
        try:    cost = resp.raw_http_response.json()["usage"].get("cost")      # Jev response
        except Exception: cost = getattr(getattr(resp, "usage", None), "cost", None)  # LLM response
        self.usd[brain] += cost or 0.0
    def __repr__(self):
        return (f"Jev: {self.calls['jev']} calls ${self.usd['jev']:.5f}  |  "
                f"LLM: {self.calls['llm']} calls ${self.usd['llm']:.5f}")
SPEND = Spend()

def ask_jev(state, questions):
    """Fast brain. state (text or JSON) + typed questions -> typed, calibrated answers."""
    r = jev.system_one(state, questions)
    SPEND.add("jev", r)
    return r

def chat(prompt, system="Be concise.", model=None):
    """Slow brain, single shot (no tools). Returns the reply text."""
    resp = client.chat.completions.create(model=model or MODEL, messages=[
        {"role": "system", "content": system}, {"role": "user", "content": prompt}])
    SPEND.add("llm", resp)
    return resp.choices[0].message.content

def show(r):
    """Print every answer in a Jev response on one line each."""
    for k, a in r.answers.items():
        if a.type == "noul":
            print(f"  {k:<18} noul   P(yes)={a.noul:.2f}")
        elif a.type == "choice":
            top = sorted(a.probabilities.items(), key=lambda kv: -kv[1])[:3]
            print(f"  {k:<18} choice {a.choice!r:<22} conf={a.confidence:.2f}  top={top}")
        else:
            level = {int(i): v for i, v in (a.legend or {}).items()}.get(round(a.score), "")
            print(f"  {k:<18} score  {a.score:.2f} -> {level!r:<18} conf={a.confidence:.2f}")

# --- Helpers used in every notebook: ask the fast brain, ask the slow brain, track spend --- class Spend: """Running tally of what OpenRouter charged, per brain (it reports exact USD per call).""" def __init__(self): self.usd, self.calls = {"jev": 0.0, "llm": 0.0}, {"jev": 0, "llm": 0} def add(self, brain, resp): self.calls[brain] += 1 try: cost = resp.raw_http_response.json()["usage"].get("cost") # Jev response except Exception: cost = getattr(getattr(resp, "usage", None), "cost", None) # LLM response self.usd[brain] += cost or 0.0 def __repr__(self): return (f"Jev: {self.calls['jev']} calls ${self.usd['jev']:.5f} | " f"LLM: {self.calls['llm']} calls ${self.usd['llm']:.5f}") SPEND = Spend() def ask_jev(state, questions): """Fast brain. state (text or JSON) + typed questions -> typed, calibrated answers.""" r = jev.system_one(state, questions) SPEND.add("jev", r) return r def chat(prompt, system="Be concise.", model=None): """Slow brain, single shot (no tools). Returns the reply text.""" resp = client.chat.completions.create(model=model or MODEL, messages=[ {"role": "system", "content": system}, {"role": "user", "content": prompt}]) SPEND.add("llm", resp) return resp.choices[0].message.content def show(r): """Print every answer in a Jev response on one line each.""" for k, a in r.answers.items(): if a.type == "noul": print(f" {k:<18} noul P(yes)={a.noul:.2f}") elif a.type == "choice": top = sorted(a.probabilities.items(), key=lambda kv: -kv[1])[:3] print(f" {k:<18} choice {a.choice!r:<22} conf={a.confidence:.2f} top={top}") else: level = {int(i): v for i, v in (a.legend or {}).items()}.get(round(a.score), "") print(f" {k:<18} score {a.score:.2f} -> {level!r:<18} conf={a.confidence:.2f}")

In [3]:

from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

def jev_map(fn, items, workers=8):
    """Run fn over items in parallel threads. Jev allows 1,200 requests/min, so 8 workers is safe."""
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(fn, items))

from pathlib import Path from concurrent.futures import ThreadPoolExecutor def jev_map(fn, items, workers=8): """Run fn over items in parallel threads. Jev allows 1,200 requests/min, so 8 workers is safe.""" with ThreadPoolExecutor(max_workers=workers) as pool: return list(pool.map(fn, items))

1. Sandboxed, read-only file tools¶

In [4]:

ROOT = Path("data/vulnerable_app").resolve()     # the sandbox: tools can't see outside it

def _safe(path):
    p = (ROOT / path).resolve()
    if p != ROOT and ROOT not in p.parents:
        raise PermissionError(f"{path!r} escapes the sandbox")
    return p

def list_dir(path="."):
    """List files under a directory inside the project (recursive)."""
    base = _safe(path)
    return sorted(str(p.relative_to(ROOT)) for p in base.rglob("*") if p.is_file())

def read_file(path):
    """Read a text file inside the project (first 8,000 chars)."""
    return _safe(path).read_text()[:8000]

print(list_dir())

ROOT = Path("data/vulnerable_app").resolve() # the sandbox: tools can't see outside it def _safe(path): p = (ROOT / path).resolve() if p != ROOT and ROOT not in p.parents: raise PermissionError(f"{path!r} escapes the sandbox") return p def list_dir(path="."): """List files under a directory inside the project (recursive).""" base = _safe(path) return sorted(str(p.relative_to(ROOT)) for p in base.rglob("*") if p.is_file()) def read_file(path): """Read a text file inside the project (first 8,000 chars).""" return _safe(path).read_text()[:8000] print(list_dir())

['README.md', 'app/admin.py', 'app/db.py', 'app/files.py', 'app/sessions.py', 'app/settings.py', 'app/views.py', 'labels.json']

2. Chunk the code with ast (code, not a model)¶

Chunking is deterministic work, so it stays in code. We split each file into one chunk per top-level function, plus a <module> chunk for top-level assignments such as constants and secrets. Each chunk is small, which suits Jev: it does best with a small, relevant state.

In [5]:

import ast

def chunks(rel_path):
    src = read_file(rel_path)
    tree, lines = ast.parse(src), src.splitlines()
    out, module_lines = [], []
    for node in tree.body:
        seg = "\n".join(lines[node.lineno - 1: node.end_lineno])
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            out.append({"id": f"{rel_path}::{node.name}", "line": node.lineno, "code": seg})
        elif isinstance(node, (ast.Assign, ast.AnnAssign)):
            module_lines.append(seg)
    if module_lines:
        out.append({"id": f"{rel_path}::<module>", "line": 1, "code": "\n".join(module_lines)})
    return out

CHUNKS = [c for f in list_dir() if f.endswith(".py") for c in chunks(f)]
print(len(CHUNKS), "chunks"); print(CHUNKS[1]["id"]); print(CHUNKS[1]["code"])

import ast def chunks(rel_path): src = read_file(rel_path) tree, lines = ast.parse(src), src.splitlines() out, module_lines = [], [] for node in tree.body: seg = "\n".join(lines[node.lineno - 1: node.end_lineno]) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): out.append({"id": f"{rel_path}::{node.name}", "line": node.lineno, "code": seg}) elif isinstance(node, (ast.Assign, ast.AnnAssign)): module_lines.append(seg) if module_lines: out.append({"id": f"{rel_path}::<module>", "line": 1, "code": "\n".join(module_lines)}) return out CHUNKS = [c for f in list_dir() if f.endswith(".py") for c in chunks(f)] print(len(CHUNKS), "chunks"); print(CHUNKS[1]["id"]); print(CHUNKS[1]["code"])

20 chunks
app/admin.py::run_backup
def run_backup(target):
    return subprocess.run(["/usr/local/bin/backup", "--target", target], check=True, capture_output=True)

3. MAP: Jev classifies every chunk¶

The CWE catalog is a Choice with a description for every option, including none. Without it, Jev would be forced to pick a vulnerability for safe code. A second Noul separates real bugs from safe look-alikes: is this really vulnerable, or does it only look similar? A chunk is flagged only when both answers agree.

In [6]:

CWES = {
    "CWE-89":  "SQL injection: untrusted input concatenated or f-string-formatted into a SQL query",
    "CWE-78":  "OS command injection: untrusted input in os.system / shell=True / command strings",
    "CWE-22":  "path traversal: untrusted filename joined to a directory without checking it stays inside",
    "CWE-798": "hardcoded credentials: secrets, API keys or passwords written in source code",
    "CWE-502": "unsafe deserialization: pickle/yaml.load on untrusted data",
    "CWE-327": "weak crypto: md5/sha1 for passwords, homemade crypto",
    "CWE-79":  "cross-site scripting: untrusted input inserted into HTML without escaping",
    "none":    "no vulnerability: parameterized queries, escaped output, validated paths, list-form subprocess, env-var secrets",
}

def scan(chunk):
    r = ask_jev({"file": chunk["id"].split("::")[0], "code": chunk["code"]}, {
        "cwe": Choice(instructions="Which vulnerability class does this code contain? Answer none if it is safe.",
                      criteria=CWES),
        "exploitable": Noul(
            instructions="Is this code really vulnerable, rather than a safe pattern that only looks similar?",
            criteria={"true": "untrusted input reaches a dangerous call unsanitized, or a real secret or weak "
                              "algorithm is in the code",
                      "false": "parameterized, escaped, validated, list-form subprocess, or secrets read from env"}),
        "severity": Score(instructions="If exploited, how bad is it?",
                          criteria=["informational", "low", "medium", "high", "critical"]),
    })
    cwe, conf = r.choices["cwe"].choice, r.choices["cwe"].confidence
    return {"id": chunk["id"], "line": chunk["line"], "cwe": cwe, "conf": conf,
            "exploitable": r.nouls["exploitable"].noul, "severity": r.scores["severity"].score,
            "flagged": cwe != "none" and r.nouls["exploitable"].noul >= 0.5}

t0 = time.perf_counter()
FINDINGS = jev_map(scan, CHUNKS)
print(f"scanned {len(CHUNKS)} chunks in {time.perf_counter() - t0:.1f} s")
for f in sorted(FINDINGS, key=lambda f: (-f["flagged"], -f["severity"])):
    mark = "!!" if f["flagged"] else "  "
    print(f"{mark} {f['id']:<35} {f['cwe']:<8} conf={f['conf']:.2f} P(exploitable)={f['exploitable']:.2f} sev={f['severity']:.1f}")

CWES = { "CWE-89": "SQL injection: untrusted input concatenated or f-string-formatted into a SQL query", "CWE-78": "OS command injection: untrusted input in os.system / shell=True / command strings", "CWE-22": "path traversal: untrusted filename joined to a directory without checking it stays inside", "CWE-798": "hardcoded credentials: secrets, API keys or passwords written in source code", "CWE-502": "unsafe deserialization: pickle/yaml.load on untrusted data", "CWE-327": "weak crypto: md5/sha1 for passwords, homemade crypto", "CWE-79": "cross-site scripting: untrusted input inserted into HTML without escaping", "none": "no vulnerability: parameterized queries, escaped output, validated paths, list-form subprocess, env-var secrets", } def scan(chunk): r = ask_jev({"file": chunk["id"].split("::")[0], "code": chunk["code"]}, { "cwe": Choice(instructions="Which vulnerability class does this code contain? Answer none if it is safe.", criteria=CWES), "exploitable": Noul( instructions="Is this code really vulnerable, rather than a safe pattern that only looks similar?", criteria={"true": "untrusted input reaches a dangerous call unsanitized, or a real secret or weak " "algorithm is in the code", "false": "parameterized, escaped, validated, list-form subprocess, or secrets read from env"}), "severity": Score(instructions="If exploited, how bad is it?", criteria=["informational", "low", "medium", "high", "critical"]), }) cwe, conf = r.choices["cwe"].choice, r.choices["cwe"].confidence return {"id": chunk["id"], "line": chunk["line"], "cwe": cwe, "conf": conf, "exploitable": r.nouls["exploitable"].noul, "severity": r.scores["severity"].score, "flagged": cwe != "none" and r.nouls["exploitable"].noul >= 0.5} t0 = time.perf_counter() FINDINGS = jev_map(scan, CHUNKS) print(f"scanned {len(CHUNKS)} chunks in {time.perf_counter() - t0:.1f} s") for f in sorted(FINDINGS, key=lambda f: (-f["flagged"], -f["severity"])): mark = "!!" if f["flagged"] else " " print(f"{mark} {f['id']:<35} {f['cwe']:<8} conf={f['conf']:.2f} P(exploitable)={f['exploitable']:.2f} sev={f['severity']:.1f}")

scanned 20 chunks in 1.3 s
!! app/sessions.py::load_session       CWE-502  conf=1.00 P(exploitable)=0.95 sev=4.0
!! app/db.py::get_user                 CWE-89   conf=1.00 P(exploitable)=0.98 sev=3.9
!! app/db.py::search_products          CWE-89   conf=1.00 P(exploitable)=0.97 sev=3.8
!! app/admin.py::ping_host             CWE-78   conf=1.00 P(exploitable)=0.90 sev=3.8
!! app/admin.py::disk_report           CWE-78   conf=1.00 P(exploitable)=0.85 sev=3.7
!! app/files.py::download              CWE-22   conf=1.00 P(exploitable)=0.85 sev=3.5
!! app/sessions.py::hash_password      CWE-327  conf=1.00 P(exploitable)=0.93 sev=3.1
!! app/settings.py::<module>           CWE-798  conf=1.00 P(exploitable)=0.60 sev=3.0
!! app/views.py::render_greeting       CWE-79   conf=1.00 P(exploitable)=0.83 sev=2.9
   app/admin.py::run_backup            none     conf=0.82 P(exploitable)=0.21 sev=3.2
   app/files.py::download_safe         CWE-22   conf=0.42 P(exploitable)=0.42 sev=2.5
   app/files.py::list_uploads          none     conf=0.75 P(exploitable)=0.15 sev=2.3
   app/sessions.py::hash_password_safe none     conf=0.95 P(exploitable)=0.16 sev=1.2
   app/settings.py::load_settings      none     conf=0.93 P(exploitable)=0.10 sev=1.0
   app/db.py::get_conn                 none     conf=0.98 P(exploitable)=0.08 sev=0.9
   app/files.py::<module>              none     conf=0.84 P(exploitable)=0.14 sev=0.7
   app/views.py::render_greeting_safe  none     conf=0.96 P(exploitable)=0.08 sev=0.4
   app/db.py::get_user_safe            none     conf=1.00 P(exploitable)=0.04 sev=0.3
   app/db.py::count_orders             none     conf=1.00 P(exploitable)=0.03 sev=0.3
   app/views.py::health                none     conf=1.00 P(exploitable)=0.03 sev=0.1

4. Score it against the planted bugs¶

In [7]:

LABELS = json.loads((ROOT / "labels.json").read_text())
truth_vuln = {k for k, v in LABELS.items() if v != "none"}
found = {f["id"] for f in FINDINGS if f["flagged"]}
right_class = {f["id"] for f in FINDINGS if f["flagged"] and LABELS.get(f["id"]) == f["cwe"]}
print(f"recall     {len(found & truth_vuln)}/{len(truth_vuln)} planted bugs flagged")
print(f"precision  {len(found & truth_vuln)}/{len(found)} flags are real")
print(f"right CWE  {len(right_class)}/{len(truth_vuln)}")
print("missed:", sorted(truth_vuln - found) or "-", "| false alarms:", sorted(found - truth_vuln) or "-")
assert len(found & truth_vuln) >= len(truth_vuln) - 2, "recall regressed"

LABELS = json.loads((ROOT / "labels.json").read_text()) truth_vuln = {k for k, v in LABELS.items() if v != "none"} found = {f["id"] for f in FINDINGS if f["flagged"]} right_class = {f["id"] for f in FINDINGS if f["flagged"] and LABELS.get(f["id"]) == f["cwe"]} print(f"recall {len(found & truth_vuln)}/{len(truth_vuln)} planted bugs flagged") print(f"precision {len(found & truth_vuln)}/{len(found)} flags are real") print(f"right CWE {len(right_class)}/{len(truth_vuln)}") print("missed:", sorted(truth_vuln - found) or "-", "| false alarms:", sorted(found - truth_vuln) or "-") assert len(found & truth_vuln) >= len(truth_vuln) - 2, "recall regressed"

recall     9/9 planted bugs flagged
precision  9/9 flags are real
right CWE  9/9
missed: - | false alarms: -

5. REDUCE: the LLM writes the report, for flagged chunks only¶

The LLM never reads the safe chunks, 11 of the 20 here. It gets only the flagged code plus Jev's verdict, and it writes the part that needs language: the exploit story and the patch.

In [8]:

flagged = sorted([f for f in FINDINGS if f["flagged"]], key=lambda f: -f["severity"])
by_id = {c["id"]: c for c in CHUNKS}

def explain(f):
    return chat(f"{f['id']} (line {f['line']}), suspected {f['cwe']}:\n```python\n{by_id[f['id']]['code']}\n```",
                system="You are an application-security reviewer. In at most 3 lines: how an attacker exploits "
                       "this, then a minimal fixed version of the code in a python block. If it is actually safe, "
                       "say 'FALSE POSITIVE' and why.")

REPORT = jev_map(explain, flagged, workers=4)
for f, text in list(zip(flagged, REPORT))[:3]:
    print(f"### {f['id']}  {f['cwe']}  severity {f['severity']:.1f}\n{text}\n")
print(SPEND)

flagged = sorted([f for f in FINDINGS if f["flagged"]], key=lambda f: -f["severity"]) by_id = {c["id"]: c for c in CHUNKS} def explain(f): return chat(f"{f['id']} (line {f['line']}), suspected {f['cwe']}:\n```python\n{by_id[f['id']]['code']}\n```", system="You are an application-security reviewer. In at most 3 lines: how an attacker exploits " "this, then a minimal fixed version of the code in a python block. If it is actually safe, " "say 'FALSE POSITIVE' and why.") REPORT = jev_map(explain, flagged, workers=4) for f, text in list(zip(flagged, REPORT))[:3]: print(f"### {f['id']} {f['cwe']} severity {f['severity']:.1f}\n{text}\n") print(SPEND)

### app/sessions.py::load_session  CWE-502  severity 4.0
An attacker can send a crafted cookie that makes `pickle.loads` execute code (potentially remote code execution).
```python
import base64, json
def load_session(cookie_value): return json.loads(base64.b64decode(cookie_value))
```

### app/db.py::get_user  CWE-89  severity 3.9
An attacker can inject SQL through `username` (for example, `' OR '1'='1`) to alter the query.
```python
def get_user(conn, username):
    cur = conn.cursor()
    cur.execute("SELECT id, email FROM users WHERE username = ?", (username,))
    return cur.fetchone()
```

### app/db.py::search_products  CWE-89  severity 3.8
An attacker can inject SQL through `term` (for example, `%' OR 1=1 --`) to alter the query.
```python
def search_products(conn, term):
    return conn.execute("SELECT * FROM products WHERE name LIKE ?", (f"%{term}%",)).fetchall()
```

Jev: 20 calls $0.00060  |  LLM: 9 calls $0.00223

6. The same capability as an agent tool¶

scan_file wraps chunk + map in one tool. The agent loop decides which files to scan and writes the prioritized summary, while Jev does the per-function work underneath.

In [9]:

def tool_spec(fn, **params):
    """Build an OpenAI tool schema from a function + {param: description} (all string params)."""
    return {"type": "function", "function": {
        "name": fn.__name__, "description": (fn.__doc__ or "").strip().split("\n")[0],
        "parameters": {"type": "object",
                       "properties": {p: {"type": "string", "description": d} for p, d in params.items()},
                       "required": list(params)}}}

def tool_spec(fn, **params): """Build an OpenAI tool schema from a function + {param: description} (all string params).""" return {"type": "function", "function": { "name": fn.__name__, "description": (fn.__doc__ or "").strip().split("\n")[0], "parameters": {"type": "object", "properties": {p: {"type": "string", "description": d} for p, d in params.items()}, "required": list(params)}}}

In [10]:

def run_agent(user_query, tools, registry, system="You are a helpful assistant. Use tools when they help. Be concise.",
              model=None, before_tool=None, check_done=None, max_iterations=8, verbose=True):
    """The agent loop from build-your-first-ai-agent, plus two Jev decision-point hooks.

    before_tool(name, args) -> None to allow, or a reason string to block      (Jev guard)
    check_done(query, answer) -> None if finished, or feedback to keep going   (Jev "am I done?" gate)
    """
    log = print if verbose else (lambda *a, **k: None)
    messages = [{"role": "system", "content": system}, {"role": "user", "content": user_query}]
    log("USER:", user_query); log("=" * 64)

    for step in range(1, max_iterations + 1):
        resp = client.chat.completions.create(model=model or MODEL, messages=messages, tools=tools)
        SPEND.add("llm", resp)
        msg = resp.choices[0].message

        # TERMINATION: no tool requested -> candidate final answer (optionally gated by Jev).
        if not msg.tool_calls:
            feedback = check_done(user_query, msg.content) if check_done else None
            if not feedback:
                log(f"[step {step}] FINAL ANSWER\n{msg.content}")
                return msg.content
            log(f"[step {step}] NOT DONE (Jev gate) -> {feedback}")
            messages += [{"role": "assistant", "content": msg.content or ""},
                         {"role": "user", "content": feedback}]
            continue

        messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [
            {"id": tc.id, "type": "function",
             "function": {"name": tc.function.name, "arguments": tc.function.arguments}}
            for tc in msg.tool_calls]})

        for tc in msg.tool_calls:
            name, args = tc.function.name, json.loads(tc.function.arguments or "{}")
            log(f"[step {step}] TOOL CALL  -> {name}({args})")
            blocked = before_tool(name, args) if before_tool else None      # <-- Jev guard
            result = {"blocked": blocked} if blocked else registry[name](**args)
            log(f"[step {step}] TOOL RESULT <- {str(result)[:300]}")
            messages.append({"role": "tool", "tool_call_id": tc.id,
                             "content": json.dumps(result, default=str)})

    return "Stopped: hit max_iterations (TTL expired)."

def run_agent(user_query, tools, registry, system="You are a helpful assistant. Use tools when they help. Be concise.", model=None, before_tool=None, check_done=None, max_iterations=8, verbose=True): """The agent loop from build-your-first-ai-agent, plus two Jev decision-point hooks. before_tool(name, args) -> None to allow, or a reason string to block (Jev guard) check_done(query, answer) -> None if finished, or feedback to keep going (Jev "am I done?" gate) """ log = print if verbose else (lambda *a, **k: None) messages = [{"role": "system", "content": system}, {"role": "user", "content": user_query}] log("USER:", user_query); log("=" * 64) for step in range(1, max_iterations + 1): resp = client.chat.completions.create(model=model or MODEL, messages=messages, tools=tools) SPEND.add("llm", resp) msg = resp.choices[0].message # TERMINATION: no tool requested -> candidate final answer (optionally gated by Jev). if not msg.tool_calls: feedback = check_done(user_query, msg.content) if check_done else None if not feedback: log(f"[step {step}] FINAL ANSWER\n{msg.content}") return msg.content log(f"[step {step}] NOT DONE (Jev gate) -> {feedback}") messages += [{"role": "assistant", "content": msg.content or ""}, {"role": "user", "content": feedback}] continue messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [ {"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in msg.tool_calls]}) for tc in msg.tool_calls: name, args = tc.function.name, json.loads(tc.function.arguments or "{}") log(f"[step {step}] TOOL CALL -> {name}({args})") blocked = before_tool(name, args) if before_tool else None # <-- Jev guard result = {"blocked": blocked} if blocked else registry[name](**args) log(f"[step {step}] TOOL RESULT <- {str(result)[:300]}") messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, default=str)}) return "Stopped: hit max_iterations (TTL expired)."

In [11]:

def scan_file(path):
    """Scan one Python file for vulnerabilities. Returns flagged functions with CWE class and severity."""
    res = jev_map(scan, chunks(path))
    return [{k: (round(v, 2) if isinstance(v, float) else v) for k, v in r.items()} for r in res if r["flagged"]]

TOOLS = [tool_spec(list_dir, path="directory inside the project, '.' for root"),
         tool_spec(read_file, path="file path inside the project"),
         tool_spec(scan_file, path="python file path inside the project")]
REGISTRY = {"list_dir": list_dir, "read_file": read_file, "scan_file": scan_file}

_ = run_agent("Audit this project for security issues. Scan every Python file, then give me a prioritized "
              "top-5 list with file, function, CWE and a one-line fix each.", TOOLS, REGISTRY,
              system="You are a security auditor. Use scan_file on each .py file; read_file only if needed.")

def scan_file(path): """Scan one Python file for vulnerabilities. Returns flagged functions with CWE class and severity.""" res = jev_map(scan, chunks(path)) return [{k: (round(v, 2) if isinstance(v, float) else v) for k, v in r.items()} for r in res if r["flagged"]] TOOLS = [tool_spec(list_dir, path="directory inside the project, '.' for root"), tool_spec(read_file, path="file path inside the project"), tool_spec(scan_file, path="python file path inside the project")] REGISTRY = {"list_dir": list_dir, "read_file": read_file, "scan_file": scan_file} _ = run_agent("Audit this project for security issues. Scan every Python file, then give me a prioritized " "top-5 list with file, function, CWE and a one-line fix each.", TOOLS, REGISTRY, system="You are a security auditor. Use scan_file on each .py file; read_file only if needed.")

USER: Audit this project for security issues. Scan every Python file, then give me a prioritized top-5 list with file, function, CWE and a one-line fix each.
================================================================

[step 1] TOOL CALL  -> list_dir({'path': '.'})
[step 1] TOOL RESULT <- ['README.md', 'app/admin.py', 'app/db.py', 'app/files.py', 'app/sessions.py', 'app/settings.py', 'app/views.py', 'labels.json']

[step 2] TOOL CALL  -> scan_file({'path': 'app/admin.py'})

[step 2] TOOL RESULT <- [{'id': 'app/admin.py::ping_host', 'line': 6, 'cwe': 'CWE-78', 'conf': 1.0, 'exploitable': 0.9, 'severity': 3.78, 'flagged': True}, {'id': 'app/admin.py::disk_report', 'line': 14, 'cwe': 'CWE-78', 'conf': 1.0, 'exploitable': 0.83, 'severity': 3.64, 'flagged': True}]
[step 2] TOOL CALL  -> scan_file({'path': 'app/db.py'})

[step 2] TOOL RESULT <- [{'id': 'app/db.py::get_user', 'line': 9, 'cwe': 'CWE-89', 'conf': 1.0, 'exploitable': 0.97, 'severity': 3.87, 'flagged': True}, {'id': 'app/db.py::search_products', 'line': 21, 'cwe': 'CWE-89', 'conf': 1.0, 'exploitable': 0.98, 'severity': 3.83, 'flagged': True}]
[step 2] TOOL CALL  -> scan_file({'path': 'app/files.py'})

[step 2] TOOL RESULT <- [{'id': 'app/files.py::download', 'line': 8, 'cwe': 'CWE-22', 'conf': 1.0, 'exploitable': 0.83, 'severity': 3.55, 'flagged': True}]
[step 2] TOOL CALL  -> scan_file({'path': 'app/sessions.py'})

[step 2] TOOL RESULT <- [{'id': 'app/sessions.py::load_session', 'line': 8, 'cwe': 'CWE-502', 'conf': 1.0, 'exploitable': 0.95, 'severity': 3.96, 'flagged': True}, {'id': 'app/sessions.py::hash_password', 'line': 12, 'cwe': 'CWE-327', 'conf': 1.0, 'exploitable': 0.93, 'severity': 3.03, 'flagged': True}]
[step 2] TOOL CALL  -> scan_file({'path': 'app/settings.py'})

[step 2] TOOL RESULT <- [{'id': 'app/settings.py::<module>', 'line': 1, 'cwe': 'CWE-798', 'conf': 1.0, 'exploitable': 0.64, 'severity': 3.13, 'flagged': True}]
[step 2] TOOL CALL  -> scan_file({'path': 'app/views.py'})

[step 2] TOOL RESULT <- [{'id': 'app/views.py::render_greeting', 'line': 5, 'cwe': 'CWE-79', 'conf': 1.0, 'exploitable': 0.82, 'severity': 2.74, 'flagged': True}]

[step 3] TOOL CALL  -> read_file({'path': 'app/sessions.py'})
[step 3] TOOL RESULT <- """Session and password helpers."""
import base64
import hashlib
import os
import pickle


def load_session(cookie_value):
    return pickle.loads(base64.b64decode(cookie_value))


def hash_password(password):
    return hashlib.md5(password.encode()).hexdigest()


def hash_password_safe(password):

[step 3] TOOL CALL  -> read_file({'path': 'app/db.py'})
[step 3] TOOL RESULT <- """Data access helpers."""
import sqlite3


def get_conn():
    return sqlite3.connect("shop.db")


def get_user(conn, username):
    cur = conn.cursor()
    cur.execute(f"SELECT id, email FROM users WHERE username = '{username}'")
    return cur.fetchone()


def get_user_safe(conn, username):
    c
[step 3] TOOL CALL  -> read_file({'path': 'app/admin.py'})
[step 3] TOOL RESULT <- """Admin utilities."""
import os
import subprocess


def ping_host(host):
    return os.system(f"ping -c 1 {host}")


def run_backup(target):
    return subprocess.run(["/usr/local/bin/backup", "--target", target], check=True, capture_output=True)


def disk_report(path):
    return subprocess.check
[step 3] TOOL CALL  -> read_file({'path': 'app/files.py'})
[step 3] TOOL RESULT <- """File download endpoints."""
import os
from pathlib import Path

BASE_DIR = "/srv/uploads"


def download(filename):
    path = os.path.join(BASE_DIR, filename)
    with open(path, "rb") as fh:
        return fh.read()


def download_safe(filename):
    base = Path(BASE_DIR).resolve()
    path = (
[step 3] TOOL CALL  -> read_file({'path': 'app/settings.py'})
[step 3] TOOL RESULT <- """Settings."""
import os

AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"
AWS_SECRET_ACCESS_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"


def load_settings():
    return {
        "database_url": os.environ.get("DATABASE_URL", "sqlite:///shop.db"),
        "debug": os.environ.get("DEBUG", "false") 
[step 3] TOOL CALL  -> read_file({'path': 'app/views.py'})
[step 3] TOOL RESULT <- """HTML views."""
import html


def render_greeting(name):
    return f"<h1>Hello {name}!</h1>"


def render_greeting_safe(name):
    return f"<h1>Hello {html.escape(name)}!</h1>"


def health():
    return {"status": "ok"}

[step 4] FINAL ANSWER
Scanned all six Python files. Prioritized by the scanner’s severity scores:

1. **`app/sessions.py` — `load_session` — CWE-502 (unsafe deserialization)**  
   Replace `pickle.loads` on cookie data with a non-executable format such as JSON, and authenticate the cookie before trusting its contents.

2. **`app/db.py` — `get_user` — CWE-89 (SQL injection)**  
   Use a parameterized query: `WHERE username = ?`, passing `(username,)` as the parameters.

3. **`app/db.py` — `search_products` — CWE-89 (SQL injection)**  
   Parameterize the `LIKE` pattern, e.g. `WHERE name LIKE ?` with `("%" + term + "%",)`.

4. **`app/admin.py` — `ping_host` — CWE-78 (OS command injection)**  
   Invoke `ping` with `subprocess.run([...])` and validate the host as an IP address or allowed hostname; avoid `os.system`.

5. **`app/admin.py` — `disk_report` — CWE-78 (OS command injection)**  
   Call `du` with an argument list and `shell=False`, and validate the requested path against an allowed directory.

In [12]:

print(SPEND)

print(SPEND)

Jev: 40 calls $0.00120  |  LLM: 13 calls $0.00308

Recap¶

• Chunk in code, map with Jev, reduce with the LLM. The LLM reads a fraction of the code, and it reads only the interesting fraction.
• A none option plus an "is it really vulnerable?" Noul is what separates get_user from get_user_safe. Two questions that must agree are more precise than one.
• The same map-reduce shape works for PR review, license scanning, PII detection in logs and policy linting, anywhere you have many chunks and few hits.

Next: 07_auto_mode_guardrails.ipynb, where Jev sits in front of every tool call and every input.