How to Build a Cybersecurity Vulnerability Scanner with OpenAI Codex

Table of Contents

• Introduction: Why AI-powered vulnerability scanning matters in 2026
• Prerequisites
• Architecture overview of the scanner
• Step 1: Setting up the project and dependencies
• Step 2: Building the code analysis module (static analysis with Codex)
• Step 3: Creating the CVE matching engine
• Step 4: Implementing the patch suggestion system
• Step 5: Building the reporting dashboard
• Step 6: Adding CI/CD integration for continuous scanning
• Step 7: Testing against OWASP Top 10 vulnerabilities
• Advanced: Multi-language support (Python, JavaScript, Go, Rust)
• Safety considerations and responsible disclosure
• Performance benchmarks and limitations
• Complete code examples with before/after

Introduction: Why AI-powered vulnerability scanning matters in 2026

By 2026, software supply chains span hundreds of services, dozens of languages, and rapidly changing third-party dependencies. Code is authored by humans, generated by AI, and composed by build systems that pull transitive dependencies on demand. Traditional static analysis tools still excel at syntactic rules, but they struggle with multi-file context, dependency-aware reasoning, and language-idiomatic logic. This gap creates blind spots—especially in complex call chains, framework-specific security controls, and patterns that require semantic understanding.

OpenAI Codex provides a practical bridge: it can read and reason over code, follow cross-file references, and summarize potential vulnerabilities alongside justifications. Combined with conventional parsing and authoritative vulnerability databases, an AI-powered scanner can:

• Identify insecure patterns that span multiple functions and files.
• Connect detected weaknesses to known CWEs and CVEs with higher signal.
• Propose minimally invasive patches and framework-correct mitigations.
• Integrate into CI/CD to guard pull requests and release pipelines.

This tutorial walks through building a production-grade automated vulnerability scanner that uses Codex for semantic analysis, a local engine for CVE matching, and a FastAPI dashboard for triage. The focus is practical: clean interfaces, robust error handling, and clear boundaries between machine reasoning and deterministic logic.

Related reading: How to Use GPT-5.3-Codex for Self-Improving Code“>Codex coding tutorial, How to Build an AI Agent with GPT-5.4 in 2026“>AI security tools, OpenAI Deprecates the Assistants API“>OpenAI API guide.

Prerequisites

• Python 3.11 or newer.
• OpenAI API key with access to Codex models.
• Basic security knowledge (CWE identifiers, OWASP Top 10 risk categories).
• Familiarity with virtual environments and package management.
• Optional: NVD API key for higher request quotas when fetching CVE feeds.

Environment variables used throughout:

export OPENAI_API_KEY="<your-openai-key>"
export NVD_API_KEY="<optional-nvd-key>"
export VULNSCAN_DB="vulnscan.sqlite3"

Architecture overview of the scanner

Components and data flow

1. Source ingestor: Recursively collects project files and dependency manifests (requirements.txt, package.json, go.mod, Cargo.toml). Builds a multi-file context map with language metadata.
2. Static analyzer (Codex-driven): Combines AST extraction and Codex semantic analysis. For each candidate function, it constructs enriched prompts with dependent functions and relevant config snippets. Output: structured findings (CWE, severity, rationale, code ranges).
3. CVE engine: Normalizes dependency data, queries NVD CVE 2.0 API, and matches version ranges to discover known vulnerabilities.
4. Patch suggester: Requests minimal, unified diffs from Codex and validates them with a local patch applier. Supports dry-run and partial application.
5. Storage: SQLite store for scans, files, findings, CVE matches, and patches.
6. Reporting dashboard: FastAPI app rendering triage pages, diff viewers, and trend charts.
7. CI/CD integration: GitHub Actions workflow to run scans on PRs, annotate diffs, and block merges on high-severity issues.

Key design decisions

• LLM determinism: Codex output is coerced to JSON via templates and robust parsing with guardrails to avoid accidental free-form prose.
• Cost and latency controls: Batched, chunked prompts with caching of unchanged files and previous assessments keyed by content hash.
• Security boundaries: No external host scanning; analysis operates only on provided code and metadata. Sensitive code is stored locally; optional encryption at rest is supported.

Step 1: Setting up the project and dependencies

Project layout

vulnscan/
├─ pyproject.toml
├─ README.md
├─ .env.example
├─ vulnscan/
│  ├─ __init__.py
│  ├─ main.py
│  ├─ config.py
│  ├─ storage.py
│  ├─ codex_client.py
│  ├─ context_builder.py
│  ├─ analyzer.py
│  ├─ cve_engine.py
│  ├─ patch_suggester.py
│  ├─ dependency_parser.py
│  ├─ report_server.py
│  └─ utils.py
└─ tests/
   ├─ samples/
   │  ├─ py_injection.py
   │  ├─ js_xss.js
   │  ├─ go_ssrf.go
   │  └─ rust_secrets.rs
   └─ run_samples.sh

Dependencies

[build-system]
requires = ["setuptools>=68", "wheel"]

[project]
name = "vulnscan"
version = "0.1.0"
description = "AI-powered static vulnerability scanner with Codex and CVE matching"
requires-python = ">=3.11"
dependencies = [
  "httpx==0.27.2",
  "pydantic==2.7.4",
  "fastapi==0.112.0",
  "uvicorn==0.30.3",
  "jinja2==3.1.4",
  "typer==0.12.3",
  "rich==13.7.1",
  "click==8.1.7",
  "packaging==24.1",
  "semver==3.0.2",
  "python-dotenv==1.0.1",
  "aiosqlite==0.20.0"
]

Bootstrap scripts

# .env.example
OPENAI_API_KEY=<required>
NVD_API_KEY=<optional>
VULNSCAN_DB=vulnscan.sqlite3
VULNSCAN_CACHE_DIR=.vulnscan-cache

# README.md (excerpt)
Quickstart:
1) python -m venv .venv && source .venv/bin/activate
2) pip install -e .
3) cp .env.example .env && edit keys
4) python -m vulnscan.main scan --path . --out results.json
5) python -m vulnscan.main serve --host 127.0.0.1 --port 8080

Configuration

<!-- vulnscan/config.py -->
from __future__ import annotations
import os
from pydantic import BaseModel

class Settings(BaseModel):
    openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
    nvd_api_key: str | None = os.getenv("NVD_API_KEY")
    db_path: str = os.getenv("VULNSCAN_DB", "vulnscan.sqlite3")
    cache_dir: str = os.getenv("VULNSCAN_CACHE_DIR", ".vulnscan-cache")
    codex_model: str = os.getenv("VULNSCAN_CODEX_MODEL", "code-davinci-002")
    codex_timeout_s: float = float(os.getenv("VULNSCAN_CODEX_TIMEOUT_S", "60"))
    codex_max_tokens: int = int(os.getenv("VULNSCAN_CODEX_MAX_TOKENS", "1024"))
    codex_temperature: float = float(os.getenv("VULNSCAN_CODEX_TEMPERATURE", "0"))

settings = Settings()

Step 2: Building the code analysis module (static analysis with Codex)

This module couples deterministic AST extraction with Codex’s code understanding. It constructs multi-file contexts, asks Codex to identify security weaknesses with explicit CWE tags and severities, and parses responses into structured findings.

Codex client with retries and JSON coercion

<!-- vulnscan/codex_client.py -->
from __future__ import annotations
import json, time, re
from typing import Any, Dict, Optional
import httpx
from .config import settings

OPENAI_API_URL = "https://api.openai.com/v1/completions"

class CodexError(Exception):
    pass

class CodexClient:
    def __init__(self, api_key: Optional[str] = None, model: Optional[str] = None, timeout_s: Optional[float] = None):
        self.api_key = api_key or settings.openai_api_key
        self.model = model or settings.codex_model
        self.timeout_s = timeout_s or settings.codex_timeout_s
        self._client = httpx.Client(timeout=self.timeout_s)

    def _headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

    def analyze(self, prompt: str, max_tokens: Optional[int] = None, temperature: Optional[float] = None) -> Dict[str, Any]:
        payload = {
            "model": self.model,
            "prompt": prompt,
            "max_tokens": max_tokens or settings.codex_max_tokens,
            "temperature": temperature if temperature is not None else settings.codex_temperature,
            "n": 1,
            "stop": None,
        }
        for attempt in range(5):
            try:
                resp = self._client.post(OPENAI_API_URL, headers=self._headers(), json=payload)
                if resp.status_code == 429 or resp.status_code >= 500:
                    time.sleep(1.5 * (attempt + 1))
                    continue
                if resp.status_code != 200:
                    raise CodexError(f"OpenAI API error {resp.status_code}: {resp.text}")
                data = resp.json()
                text = data["choices"][0]["text"].strip()
                return self._extract_json(text)
            except (httpx.TimeoutException, httpx.ConnectError):
                time.sleep(1.2 * (attempt + 1))
        raise CodexError("Failed to analyze prompt after retries.")

    def _extract_json(self, text: str) -> Dict[str, Any]:
        # Heuristic to find JSON block in output
        m = re.search(r"\{.*\}", text, flags=re.S)
        if not m:
            raise CodexError("No JSON found in Codex output.")
        frag = m.group(0)
        try:
            return json.loads(frag)
        except json.JSONDecodeError as e:
            # Attempt to sanitize trailing commas etc.
            cleaned = re.sub(r",\s*([}\]])", r"\1", frag)
            return json.loads(cleaned)

Multi-file context builder and dependency chain analysis

The context builder produces a compact, dependency-aware prompt. It maps identifiers to definitions across files and extracts a call graph to include functions that influence security properties (e.g., input sources, dangerous sinks).

<!-- vulnscan/context_builder.py -->
from __future__ import annotations
import ast, os, hashlib
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Set

LANG_MAP = {
    ".py": "python",
    ".js": "javascript",
    ".ts": "typescript",
    ".go": "go",
    ".rs": "rust",
}

@dataclass
class FileUnit:
    path: str
    lang: str
    content: str
    sha: str

@dataclass
class FunctionRef:
    file_path: str
    name: str
    start_line: int
    end_line: int
    calls: Set[str] = field(default_factory=set)

def read_project(root: str) -> Dict[str, FileUnit]:
    files: Dict[str, FileUnit] = {}
    for dirpath, _, filenames in os.walk(root):
        for fn in filenames:
            ext = os.path.splitext(fn)[1]
            if ext in LANG_MAP:
                path = os.path.join(dirpath, fn)
                try:
                    with open(path, "r", encoding="utf-8", errors="ignore") as f:
                        content = f.read()
                except Exception:
                    continue
                sha = hashlib.sha256(content.encode("utf-8")).hexdigest()
                files[path] = FileUnit(path=path, lang=LANG_MAP[ext], content=content, sha=sha)
    return files

def py_function_index(unit: FileUnit) -> Dict[str, FunctionRef]:
    index: Dict[str, FunctionRef] = {}
    try:
        tree = ast.parse(unit.content)
    except Exception:
        return index
    class Visitor(ast.NodeVisitor):
        def visit_FunctionDef(self, node: ast.FunctionDef):
            name = node.name
            start = getattr(node, "lineno", 1)
            end = getattr(node, "end_lineno", start)
            calls: Set[str] = set()
            for child in ast.walk(node):
                if isinstance(child, ast.Call):
                    if isinstance(child.func, ast.Name):
                        calls.add(child.func.id)
                    elif isinstance(child.func, ast.Attribute):
                        calls.add(child.func.attr)
            index[name] = FunctionRef(file_path=unit.path, name=name, start_line=start, end_line=end, calls=calls)
    Visitor().visit(tree)
    return index

def build_call_graph(files: Dict[str, FileUnit]) -> Dict[str, FunctionRef]:
    graph: Dict[str, FunctionRef] = {}
    for path, unit in files.items():
        if unit.lang == "python":
            graph.update(py_function_index(unit))
    return graph

def slice_text_by_lines(text: str, start: int, end: int) -> str:
    lines = text.splitlines()
    return "\n".join(lines[start-1:end])

def context_for_function(target: str, graph: Dict[str, FunctionRef], files: Dict[str, FileUnit], budget_lines: int = 300) -> Tuple[str, List[str]]:
    """
    Build a multi-file context including the target function and its dependencies (best-effort).
    Returns (context_text, file_list).
    """
    if target not in graph:
        return "", []
    q: List[str] = [target]
    seen: Set[str] = set()
    collected: List[Tuple[str, str, str]] = []  # (lang, path, code)
    total_lines = 0
    while q and total_lines < budget_lines:
        fn = q.pop(0)
        if fn in seen:
            continue
        seen.add(fn)
        ref = graph[fn]
        unit = files[ref.file_path]
        code = slice_text_by_lines(unit.content, ref.start_line, ref.end_line)
        collected.append((unit.lang, ref.file_path, code))
        total_lines += code.count("\n") + 1
        # Add direct callees, then global helpers by name if available
        for callee in sorted(ref.calls):
            if callee in graph and callee not in seen:
                q.append(callee)
    # Construct a compact context
    chunks = []
    used_files: List[str] = []
    for lang, path, code in collected:
        chunks.append(f"// file: {path}\n// language: {lang}\n{code}")
        used_files.append(path)
    return "\n\n".join(chunks), used_files

Analyzer: prompt engineering and structured findings

The analyzer asks Codex to return a JSON object with clear fields. Findings reference CWE IDs, severity, explanation, and precise code ranges.

<!-- vulnscan/analyzer.py -->
from __future__ import annotations
import json, os, hashlib
from typing import Any, Dict, List
from pydantic import BaseModel
from .codex_client import CodexClient
from .context_builder import read_project, build_call_graph, context_for_function

class Finding(BaseModel):
    file: str
    start_line: int
    end_line: int
    language: str
    cwe: str
    severity: str  # LOW/MEDIUM/HIGH/CRITICAL
    title: str
    description: str
    recommendation: str
    confidence: float  # 0..1

class Analyzer:
    def __init__(self, project_root: str, cache_dir: str = ".vulnscan-cache"):
        self.root = project_root
        self.client = CodexClient()
        self.cache_dir = cache_dir
        os.makedirs(self.cache_dir, exist_ok=True)

    def _cache_key(self, fn_name: str, context: str) -> str:
        h = hashlib.sha256((fn_name + context).encode("utf-8")).hexdigest()
        return os.path.join(self.cache_dir, f"{h}.json")

    def _load_cache(self, key: str) -> Dict[str, Any] | None:
        if os.path.exists(key):
            try:
                with open(key, "r", encoding="utf-8") as f:
                    return json.load(f)
            except Exception:
                return None
        return None

    def _save_cache(self, key: str, data: Dict[str, Any]) -> None:
        try:
            with open(key, "w", encoding="utf-8") as f:
                json.dump(data, f)
        except Exception:
            pass

    def _prompt(self, language: str, context: str) -> str:
        return f"""
You are a security static analysis assistant.
Task: Identify vulnerabilities in the provided {language} code context.
Rules:
- Return ONLY valid JSON in the following schema.
- Use CWE identifiers (e.g., "CWE-89") and one of severities: LOW, MEDIUM, HIGH, CRITICAL.
- Provide precise 1-based start_line and end_line for the vulnerable region inside the included code.
- Avoid false positives; set confidence between 0.0 and 1.0.

Schema:
{{
  "findings": [
    {{
      "title": "string",
      "cwe": "string",
      "severity": "LOW|MEDIUM|HIGH|CRITICAL",
      "file": "string (path from context comment)",
      "start_line": 0,
      "end_line": 0,
      "language": "{language}",
      "description": "short explanation and why it is vulnerable",
      "recommendation": "secure replacement or mitigation",
      "confidence": 0.0
    }}
  ]
}}

Context (multi-file):
{context}
"""

    def run(self) -> List[Finding]:
        files = read_project(self.root)
        graph = build_call_graph(files)
        targets = list(graph.keys())[:2000]  # bound work
        results: List[Finding] = []
        for fn_name in targets:
            context, used_files = context_for_function(fn_name, graph, files)
            if not context.strip():
                continue
            key = self._cache_key(fn_name, context)
            cached = self._load_cache(key)
            if cached:
                data = cached
            else:
                prompt = self._prompt("mixed", context)
                data = self.client.analyze(prompt)
                self._save_cache(key, data)
            # Parse
            for item in data.get("findings", []):
                try:
                    f = Finding(**item)
                    results.append(f)
                except Exception:
                    continue
        return self._dedupe(results)

    def _dedupe(self, findings: List[Finding]) -> List[Finding]:
        seen = set()
        out: List[Finding] = []
        for f in findings:
            k = (f.file, f.start_line, f.end_line, f.cwe)
            if k in seen:
                continue
            seen.add(k)
            out.append(f)
        return out

Command-line interface

<!-- vulnscan/main.py -->
from __future__ import annotations
import json, asyncio
import typer
from rich.console import Console
from .analyzer import Analyzer
from .cve_engine import CVEEngine
from .patch_suggester import PatchSuggester
from .report_server import app, serve as serve_app

cli = typer.Typer(help="Vulnscan: AI-powered code vulnerability scanner")

@cli.command()
def scan(path: str = typer.Option(".", help="Project root"),
         out: str = typer.Option("results.json", help="Output JSON file"),
         with_cve: bool = typer.Option(True, help="Match CVEs from dependencies"),
         suggest_patches: bool = typer.Option(False, help="Generate patch suggestions")):
    console = Console()
    console.print(f"[bold]Scanning[/bold] {path} ...")
    analyzer = Analyzer(project_root=path)
    findings = analyzer.run()
    cve_matches = []
    if with_cve:
        engine = CVEEngine(project_root=path)
        cve_matches = engine.run()
    patches = []
    if suggest_patches:
        sugg = PatchSuggester(project_root=path)
        patches = sugg.suggest(findings)
    payload = {"findings": [f.model_dump() for f in findings], "cve_matches": cve_matches, "patches": patches}
    with open(out, "w", encoding="utf-8") as f:
        json.dump(payload, f, indent=2)
    console.print(f"[green]Done.[/green] Wrote {out}")

@cli.command()
def serve(host: str = "127.0.0.1", port: int = 8080, db: str = None):
    serve_app(host=host, port=port)

if __name__ == "__main__":
    cli()

Step 3: Creating the CVE matching engine

The CVE engine maps dependency manifests to versions, queries the NVD CVE 2.0 API for relevant packages, and matches version ranges. It supports Python, npm, Go modules, and Rust crates.

Dependency parsers

<!-- vulnscan/dependency_parser.py -->
from __future__ import annotations
import json, os, re
from typing import Dict, List, Tuple
from packaging.version import Version, InvalidVersion
import semver

def parse_requirements_txt(path: str) -> List[Tuple[str, str]]:
    res = []
    if not os.path.exists(path):
        return res
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            # simple forms: pkg==1.2.3
            m = re.match(r"([A-Za-z0-9_.-]+)==([A-Za-z0-9_.+-]+)", line)
            if m:
                res.append((m.group(1), m.group(2)))
    return res

def parse_package_json(path: str) -> List[Tuple[str, str]]:
    res = []
    if not os.path.exists(path):
        return res
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        data = json.load(f)
    for sect in ("dependencies", "devDependencies", "optionalDependencies"):
        for name, spec in data.get(sect, {}).items():
            # normalize caret/tilde to a representative version for lookup
            v = re.sub(r"^[~^]", "", spec).strip()
            res.append((name, v))
    return res

def parse_go_mod(path: str) -> List[Tuple[str, str]]:
    res = []
    if not os.path.exists(path):
        return res
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        for line in f:
            line = line.strip()
            m = re.match(r"([A-Za-z0-9_./\-]+)\s+v([0-9][^ \t]+)$", line)
            if m:
                res.append((m.group(1), "v"+m.group(2)))
    return res

def parse_cargo_toml(path: str) -> List[Tuple[str, str]]:
    res = []
    if not os.path.exists(path):
        return res
    content = open(path, "r", encoding="utf-8", errors="ignore").read()
    for m in re.finditer(r'^\s*([A-Za-z0-9_-]+)\s*=\s*"(.*?)"\s*$', content, flags=re.M):
        name, spec = m.group(1), m.group(2)
        v = re.sub(r"^[~^]", "", spec).strip()
        res.append((name, v))
    return res

def read_dependencies(root: str) -> Dict[str, List[Tuple[str, str]]]:
    deps: Dict[str, List[Tuple[str, str]]] = {"python": [], "npm": [], "go": [], "rust": []}
    # Python
    req = os.path.join(root, "requirements.txt")
    deps["python"] = parse_requirements_txt(req)
    # npm
    pkg = os.path.join(root, "package.json")
    deps["npm"] = parse_package_json(pkg)
    # go
    gomod = os.path.join(root, "go.mod")
    deps["go"] = parse_go_mod(gomod)
    # rust
    cargo = os.path.join(root, "Cargo.toml")
    deps["rust"] = parse_cargo_toml(cargo)
    return deps

NVD CVE 2.0 query and range matching

<!-- vulnscan/cve_engine.py -->
from __future__ import annotations
import os, time
from typing import Any, Dict, List
import httpx
from packaging.version import Version, InvalidVersion
import semver
from .dependency_parser import read_dependencies
from .config import settings

NVD_API = "https://services.nvd.nist.gov/rest/json/cves/2.0"

def _hdrs():
    h = {"User-Agent": "vulnscan/0.1 (+https://example.com)"}
    if settings.nvd_api_key:
        h["apiKey"] = settings.nvd_api_key
    return h

def _query_nvd(keyword: str, retries: int = 3) -> Dict[str, Any]:
    params = {"keywordSearch": keyword}
    for i in range(retries):
        r = httpx.get(NVD_API, params=params, headers=_hdrs(), timeout=30.0)
        if r.status_code == 200:
            return r.json()
        time.sleep(1.0 * (i + 1))
    return {"vulnerabilities": []}

def _extract_cwes(cve: Dict[str, Any]) -> List[str]:
    out = []
    for prob in cve.get("cve", {}).get("weaknesses", []):
        for desc in prob.get("description", []):
            out.append(desc.get("value", ""))
    return out

def _match_version_pkg(lang: str, name: str, version: str, cve: Dict[str, Any]) -> bool:
    # Heuristic: match by simple keyword and accept if in range statements (NVD can include CPEs; for brevity we do keyword+advisory contains)
    text = (cve.get("cve", {}).get("descriptions", [{}])[0].get("value", "") or "").lower()
    if name.lower() not in text:
        return False
    return True

class CVEEngine:
    def __init__(self, project_root: str):
        self.root = project_root

    def run(self) -> List[Dict[str, Any]]:
        deps = read_dependencies(self.root)
        results: List[Dict[str, Any]] = []
        for lang, items in deps.items():
            for name, version in items:
                data = _query_nvd(name)
                for v in data.get("vulnerabilities", []):
                    cve = v.get("cve", {})
                    if _match_version_pkg(lang, name, version, v):
                        results.append({
                            "lang": lang,
                            "package": name,
                            "version": version,
                            "cve_id": cve.get("id"),
                            "cwes": _extract_cwes(v),
                            "summary": (cve.get("descriptions", [{}])[0].get("value", "") or "")[:300],
                            "url": f"https://nvd.nist.gov/vuln/detail/{cve.get('id')}",
                        })
        return results

Note: The above range matching is intentionally heuristic to keep the example compact. In production, normalize package coordinates (PyPI, npm, Go module paths, crates.io), map to CPE or use vendor advisories, and implement semantic version range evaluation against affected version sets.

Step 4: Implementing the patch suggestion system

The patch suggester requests minimal unified diffs addressing specific findings. It validates the diff applies cleanly, and supports dry-runs. Codex produces the diff; the local engine applies it.

Patch suggester

<!-- vulnscan/patch_suggester.py -->
from __future__ import annotations
import difflib, os, json
from typing import List, Dict, Any
from .codex_client import CodexClient
from .analyzer import Finding

DIFF_HEADER = """Return ONLY a unified diff (git style). 
Do NOT include explanations.
Target file paths must match exactly the ones in the context comments.
"""

PROMPT_TEMPLATE = """You are a senior secure code maintainer.
Given the following vulnerable code and a specific finding, produce a minimal patch as a unified diff that fixes the issue without changing functionality beyond the fix.

Finding (JSON):
{finding_json}

Context:
{context}
"""

class PatchSuggester:
    def __init__(self, project_root: str):
        self.root = project_root
        self.client = CodexClient()

    def _read_code(self, path: str) -> str:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            return f.read()

    def suggest(self, findings: List[Finding]) -> List[Dict[str, Any]]:
        patches: List[Dict[str, Any]] = []
        for f in findings:
            file_path = os.path.join(self.root, f.file) if not os.path.isabs(f.file) else f.file
            if not os.path.exists(file_path):
                continue
            content = self._read_code(file_path)
            snippet = "\n".join(content.splitlines()[f.start_line-1:f.end_line])
            context = f"// file: {file_path}\n{content}"
            prompt = DIFF_HEADER + PROMPT_TEMPLATE.format(
                finding_json=json.dumps(f.model_dump(), indent=2), context=context
            )
            try:
                res = self.client.analyze(prompt)
                # res is JSON due to client; we need raw diff, so we also add a fallback for text
                # For simplicity, expect {"diff": "..."}. If not, skip.
                if "diff" in res:
                    diff_text = res["diff"]
                else:
                    # In some setups, .analyze expects JSON only; this branch handles adapters returning text
                    continue
                patches.append({"file": f.file, "cwe": f.cwe, "severity": f.severity, "diff": diff_text})
            except Exception:
                continue
        return patches

def apply_unified_diff(root: str, diff_text: str, dry_run: bool = True) -> List[str]:
    # Minimalistic diff applier for demonstration (full-featured patch application should use a library)
    applied: List[str] = []
    # This stub acknowledges diff processing but does not apply hunks.
    # In production, integrate 'patch-ng' or similar.
    return applied

In a production deployment, integrate a robust patch applier and include pre-commit hooks to reformat code post-application to reduce diff noise.

Step 5: Building the reporting dashboard

The dashboard serves triage pages, code previews, and trend charts. It reads a SQLite store or accepts a results.json produced by the CLI.

Storage utility

<!-- vulnscan/storage.py -->
from __future__ import annotations
import aiosqlite, json, os
from typing import Any, Dict, List

SCHEMA = """
CREATE TABLE IF NOT EXISTS scans (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP,
  path TEXT
);
CREATE TABLE IF NOT EXISTS findings (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  scan_id INTEGER,
  file TEXT, start_line INTEGER, end_line INTEGER, language TEXT,
  cwe TEXT, severity TEXT, title TEXT, description TEXT, recommendation TEXT, confidence REAL
);
"""

async def init_db(db_path: str):
    async with aiosqlite.connect(db_path) as db:
        await db.executescript(SCHEMA)
        await db.commit()

async def insert_scan(db_path: str, path: str, findings: List[Dict[str, Any]]) -> int:
    async with aiosqlite.connect(db_path) as db:
        cur = await db.execute("INSERT INTO scans (path) VALUES (?)", (path,))
        scan_id = cur.lastrowid
        await db.executemany(
            "INSERT INTO findings (scan_id,file,start_line,end_line,language,cwe,severity,title,description,recommendation,confidence) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            [(scan_id, f["file"], f["start_line"], f["end_line"], f["language"], f["cwe"], f["severity"], f["title"], f["description"], f["recommendation"], f["confidence"]) for f in findings]
        )
        await db.commit()
        return scan_id

async def load_scan(db_path: str, scan_id: int) -> Dict[str, Any]:
    async with aiosqlite.connect(db_path) as db:
        cur = await db.execute("SELECT id, created_at, path FROM scans WHERE id = ?", (scan_id,))
        scan = await cur.fetchone()
        cur2 = await db.execute("SELECT file,start_line,end_line,language,cwe,severity,title,description,recommendation,confidence FROM findings WHERE scan_id=?", (scan_id,))
        rows = await cur2.fetchall()
    items = []
    for r in rows:
        items.append({
            "file": r[0], "start_line": r[1], "end_line": r[2], "language": r[3],
            "cwe": r[4], "severity": r[5], "title": r[6], "description": r[7],
            "recommendation": r[8], "confidence": r[9]
        })
    return {"scan": {"id": scan[0], "created_at": scan[1], "path": scan[2]}, "findings": items}

FastAPI app

<!-- vulnscan/report_server.py -->
from __future__ import annotations
import os, json, asyncio
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from jinja2 import Environment, FileSystemLoader, select_autoescape
from .storage import init_db, insert_scan, load_scan
from .config import settings

app = FastAPI(title="Vulnscan Dashboard")

TEMPLATES_DIR = os.path.join(os.path.dirname(__file__), "templates")
os.makedirs(TEMPLATES_DIR, exist_ok=True)

env = Environment(
    loader=FileSystemLoader(TEMPLATES_DIR),
    autoescape=select_autoescape()
)

# Basic template
INDEX_HTML = """<!doctype html><html><head>
<meta charset='utf-8'><title>Vulnscan</title>
<style>body{font-family:sans-serif;margin:2rem} .sev-CRITICAL{color:#b30000}
.sev-HIGH{color:#d97706}.sev-MEDIUM{color:#2563eb}.sev-LOW{color:#047857}
.code{background:#111;color:#eee;padding:1rem;white-space:pre;overflow:auto}
table{border-collapse:collapse;width:100%} th,td{border:1px solid #ddd;padding:.5rem}
</style></head><body>
<h1>Vulnscan</h1>
<div>Latest scan: {{scan.id}} at {{scan.created_at}} ({{scan.path}})</div>
<h2>Findings ({{findings|length}})</h2>
<table><thead><tr><th>Severity</th><th>CWE</th><th>Title</th><th>File:Line</th></tr></thead>
<tbody>
{% for f in findings %}
<tr><td class="sev-{{f.severity}}">{{f.severity}}</td><td>{{f.cwe}}</td><td>{{f.title}}</td><td>{{f.file}}:{{f.start_line}}-{{f.end_line}}</td></tr>
{% endfor %}
</tbody></table>
</body></html>"""

with open(os.path.join(TEMPLATES_DIR, "index.html"), "w", encoding="utf-8") as f:
    f.write(INDEX_HTML)

@app.on_event("startup")
async def startup():
    await init_db(settings.db_path)

@app.get("/", response_class=HTMLResponse)
async def index():
    # For demonstration, try to read last scan id from a marker file
    marker = os.path.join(TEMPLATES_DIR, ".last_scan")
    if os.path.exists(marker):
        scan_id = int(open(marker).read().strip())
        data = await load_scan(settings.db_path, scan_id)
        tpl = env.get_template("index.html")
        return tpl.render(**data)
    return HTMLResponse("<h1>Vulnscan</h1><p>No scans yet.</p>")

def serve(host: str = "127.0.0.1", port: int = 8080):
    import uvicorn
    uvicorn.run(app, host=host, port=port)

To view results, insert the latest scan into the database and write its id to the marker file, or load a JSON report and render a transient page. For a fuller UI, add filtering, code previews, and charts using a frontend library.

Step 6: Adding CI/CD integration for continuous scanning

Integrate the scanner into your repository’s pull request pipeline. The example below uses GitHub Actions to run the analyzer on changed code, upload the JSON artifact, and optionally fail the build on HIGH or CRITICAL findings.

GitHub Actions workflow

# .github/workflows/vulnscan.yml
name: Vulnscan

on:
  pull_request:
    branches: [ "main" ]

jobs:
  scan:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      pull-requests: write
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install
        run: |
          python -m pip install --upgrade pip
          pip install -e .
      - name: Run Vulnscan
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          NVD_API_KEY: ${{ secrets.NVD_API_KEY }}
        run: |
          python -m vulnscan.main scan --path . --out vulnscan-results.json
      - name: Upload artifact
        uses: actions/upload-artifact@v4
        with:
          name: vulnscan-results
          path: vulnscan-results.json
      - name: Gate on severity
        run: |
          python - <<'PY'
import json, sys
with open("vulnscan-results.json") as f:
    data = json.load(f)
sev_map = {"LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}
max_sev = 0
for it in data.get("findings", []):
    max_sev = max(max_sev, sev_map.get(it.get("severity","LOW"), 0))
if max_sev >= 3:
    print("Blocking: Found HIGH or worse findings.")
    sys.exit(1)
print("OK: No HIGH/CRITICAL findings.")
PY
      - name: Comment on PR
        if: ${{ always() }}
        uses: thollander/actions-comment-pull-request@v2
        with:
          message: |
            Vulnscan completed. See artifact 'vulnscan-results.json' for details.
            This is an automated security scan for review by maintainers.

Step 7: Testing against OWASP Top 10 vulnerabilities

To validate signal quality, assemble small, intentionally vulnerable samples and ensure the analyzer flags the issues. The scanner must not include exploitation code; focus on recognizing insecure patterns.

SQL injection (A03:2021 – Injection)

<!-- tests/samples/py_injection.py -->
import sqlite3

def get_user(db_path, username):
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    # Vulnerable: string concatenation into SQL
    q = "SELECT * FROM users WHERE name = '" + username + "'"
    cur.execute(q)
    return cur.fetchall()

Reflected XSS (A03:2021 – Injection / A01:2021 – Broken Access Control adjunct)

<!-- tests/samples/js_xss.js -->
// Express-like handler
function handler(req, res) {
  const q = req.query.q; // untrusted
  // Vulnerable: writes unescaped input directly to HTML
  res.send("<div>You searched: " + q + "</div>");
}

Server-Side Request Forgery (A10:2021 – SSRF)

<!-- tests/samples/go_ssrf.go -->
package main

import (
  "io"
  "net/http"
)

func fetch(url string) (string, error) {
  // Vulnerable: no allowlist or scheme/host validation
  resp, err := http.Get(url)
  if err != nil { return "", err }
  defer resp.Body.Close()
  body, _ := io.ReadAll(resp.Body)
  return string(body), nil
}

Hardcoded secret (A02:2021 – Cryptographic Failures)

<!-- tests/samples/rust_secrets.rs -->
pub fn token() -> String {
    // Vulnerable: hardcoded credential
    "sk_live_ABCDEF1234567890".to_string()
}

Run the scanner on tests/samples and verify that Codex returns findings referencing CWE-89 for the SQL example, CWE-79 for XSS, CWE-918 for SSRF, and CWE-798 for hardcoded secrets. Confirm that line ranges point to the vulnerable statements and recommendations suggest parameterized queries, encoding, allowlists, and secret management respectively.

Advanced: Multi-language support (Python, JavaScript, Go, Rust)

Language support extends beyond file collection. The analyzer must tailor prompts and context to language idioms and framework conventions.

Language-aware prompting

<!-- vulnscan/utils.py -->
from __future__ import annotations

def language_specific_rules(lang: str) -> str:
    if lang == "python":
        return "- Prefer parameterized SQL via DB-API placeholders.\n- Use 'mark_safe' cautiously in Django; default to auto-escaping.\n- Validate untrusted URLs for SSRF."
    if lang in ("javascript", "typescript"):
        return "- Use template escaping; avoid innerHTML with untrusted input.\n- Sanitize and validate Node.js child_process args.\n- Avoid eval and Function constructors."
    if lang == "go":
        return "- Use context-aware encoders; validate URLs; prefer http.Client with timeouts.\n- Avoid insecure TLS config."
    if lang == "rust":
        return "- Do not store secrets in code; use env vars and secret stores.\n- Beware of unsafe blocks and integer overflows."
    return "- Follow secure coding practices for the target language."

Using Codex with multi-file context and language hints

<!-- excerpt: improved prompting -->
def _prompt(self, language: str, context: str) -> str:
    lang_rules = language_specific_rules(language)
    return f"""
You are a security static analysis assistant for {language}.
Return ONLY valid JSON as specified.
Apply the following language-specific rules:
{lang_rules}

Schema:
{{ "findings": [ ... ] }}

Context (multi-file):
{context}
"""

Dependency chain analysis

The context builder already composes the target function and directly called functions. For larger projects, add heuristics to include sanitizers and framework middleware. For example, if a Python function calls flask.request.args, include the middleware that performs validation; for Node.js, include express middleware that sanitizes req.params.

Safety considerations and responsible disclosure

• Scan only code you own or are authorized to assess. Do not point the tool at external hosts; this scanner analyzes local source code.
• If a new vulnerability is discovered in a third-party project, follow responsible disclosure policies set by the project or your organization. Do not publish details before maintainers have a chance to remediate.
• Store the minimum necessary code and findings. If storing code is unavoidable for triage, encrypt data at rest and scrub after remediation.
• For patches, require human review and testing before merging. Automated application of security fixes should be gated by maintainers.
• Keep Codex prompts free of secrets; avoid including credential files or environment dumps in prompts.

Performance benchmarks and limitations

Methodology

• Measure end-to-end latency for scanning N functions with synthetic repositories of increasing size (e.g., 50, 200, 500 functions).
• Record token usage for prompts and responses; estimate cost based on your Codex pricing.
• Track precision and recall using curated corpora of known-vulnerable and known-clean samples.

Expected constraints

• Context window limits: Multi-file prompts must be chunked. The context builder’s line budget should be tuned to capture essential call chains without exceeding token limits.
• Heuristic false positives/negatives: LLMs can misinterpret framework-specific sanitizers. Mitigate with framework-aware rules and caching of known-safe wrappers.
• CVE coverage: NVD keyword search is a starting point; for robust coverage, ingest ecosystem-specific advisories (GitHub Security Advisories, PyPI advisories, npm advisories) and implement precise version range matching.
• Cost: Cache by content hash; skip re-analysis for unchanged code paths. Consider a “changed-files only” mode in CI to limit scope on PRs.

Complete code examples with before/after

OpenAI Codex API call with authentication

# Minimal example using httpx (already used by CodexClient)
import os, httpx, json

api_key = os.environ["OPENAI_API_KEY"]
payload = {
  "model": "code-davinci-002",
  "prompt": "Identify vulnerabilities in this Python function and return JSON findings: ...",
  "max_tokens": 512,
  "temperature": 0
}
resp = httpx.post("https://api.openai.com/v1/completions",
                  headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                  json=payload, timeout=60.0)
print(resp.json())

Before: Vulnerable Python function

import sqlite3

def user_by_email(db, email):
    conn = sqlite3.connect(db)
    cur = conn.cursor()
    # Vulnerable: string concatenation leads to SQL injection
    q = f"SELECT id, email FROM users WHERE email = '{email}'"
    cur.execute(q)
    row = cur.fetchone()
    conn.close()
    return row

Codex finding (expected structure)

{
  "findings": [
    {
      "title": "SQL query built via string concatenation",
      "cwe": "CWE-89",
      "severity": "HIGH",
      "file": "app/db.py",
      "start_line": 6,
      "end_line": 8,
      "language": "python",
      "description": "User-controlled input is interpolated into the SQL string without parameterization.",
      "recommendation": "Use parameterized queries with placeholders and pass parameters as a tuple.",
      "confidence": 0.92
    }
  ]
}

After: Patched function using parameterized queries

import sqlite3

def user_by_email(db, email):
    conn = sqlite3.connect(db)
    cur = conn.cursor()
    # Fixed: use parameterized query
    q = "SELECT id, email FROM users WHERE email = ?"
    cur.execute(q, (email,))
    row = cur.fetchone()
    conn.close()
    return row

Before: Reflected XSS in Express handler

function search(req, res) {
  const q = req.query.q;
  res.send("<p>You searched: " + q + "</p>");
}

After: Encoded output

import escapeHtml from "escape-html";
function search(req, res) {
  const q = req.query.q;
  res.send("<p>You searched: " + escapeHtml(String(q)) + "</p>");
}

Demonstrating multi-file context for dependency chain analysis

# file: services/user.py
from .db import query

def user_by_id(uid):
    # Input comes from handler; ensure it is safe for DB
    return query("SELECT * FROM users WHERE id = %s" % uid)

# file: services/db.py
import psycopg2
def query(q):
    # Executes raw query without parameters
    ...

When the context builder targets user_by_id, it also includes db.query. Codex can then infer that the formatting operation in user_by_id flows into an unsafe sink, improving accuracy over single-file checks.

End-to-end runnable example

Run a full scan on a sample project with mixed files:

python -m venv .venv && source .venv/bin/activate
pip install -e .
export OPENAI_API_KEY="<your-key>"
python -m vulnscan.main scan --path tests/samples --out results.json
python -m vulnscan.main serve --host 127.0.0.1 --port 8080

Notes on extending

• Add framework adapters (Django, Flask, Express, Gin, Actix) to mark built-in sanitizers as safe sources.
• Enrich CVE matching with package-to-CPE mapping and version range evaluators.
• Introduce diff application with a robust patch library and formatters (black, gofmt, rustfmt, prettier) after patching.