AI Code Review & Fuzzing

# AI-assisted code review framework
# Works with: current OpenAI / Claude models, open-weight coders
# (Qwen3-Coder, DeepSeek V4), and local models via Ollama

import openai
from pathlib import Path

SECURITY_AUDIT_PROMPT = """You are an expert security code reviewer. Analyse the following code for:

1. **Injection vulnerabilities**: SQL injection, command injection, LDAP injection, XSS
2. **Authentication/Authorisation flaws**: Broken auth, IDOR, privilege escalation
3. **Cryptographic issues**: Weak algorithms, hardcoded keys, improper random
4. **Memory safety**: Buffer overflows, use-after-free (for C/C++/Rust unsafe)
5. **Deserialization**: Unsafe deserialization of user input
6. **SSRF / Path traversal**: Server-side request forgery, directory traversal
7. **Race conditions**: TOCTOU, missing locks on shared state
8. **Secrets exposure**: API keys, tokens, passwords in source
9. **Dependency risks**: Known-vulnerable library patterns

For each finding:
- Severity: CRITICAL / HIGH / MEDIUM / LOW / INFO
- Vulnerable code: Quote the exact line(s)
- Impact: What an attacker can achieve
- CWE: Map to CWE ID
- Remediation: Specific fix with code example

If no vulnerabilities are found, state that explicitly. Do not invent findings."""

def audit_file(filepath: str, model: str = "gpt-5") -> str:
    """Audit a single source file for security vulnerabilities."""
    code = Path(filepath).read_text()
    
    response = openai.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": SECURITY_AUDIT_PROMPT},
            {"role": "user", "content": f"File: {filepath}\n\n```\n{code}\n```"}
        ],
        temperature=0.1  # Low temp → more precise, fewer hallucinations
    )
    return response.choices[0].message.content

def audit_directory(directory: str, extensions: list = None):
    """Audit all matching files in a directory tree."""
    if extensions is None:
        extensions = ['.py', '.js', '.ts', '.go', '.c', '.cpp', '.java', '.rs', '.php', '.rb']
    
    findings = []
    for path in Path(directory).rglob('*'):
        if path.suffix in extensions and path.is_file():
            print(f"Auditing: {path}")
            result = audit_file(str(path))
            if "no vulnerabilities" not in result.lower():
                findings.append({"file": str(path), "findings": result})
    
    return findings

# Example: audit a Python web app
results = audit_directory("./target-app/src/", ['.py'])
for r in results:
    print(f"\n{'='*60}")
    print(f"FILE: {r['file']}")
    print(r['findings'])

# Step 1: Run Semgrep to get candidate findings
pip install semgrep

# Scan with security-focused rules
semgrep scan --config=p/security-audit \
             --config=p/owasp-top-ten \
             --config=p/cwe-top-25 \
             --json --output findings.json \
             ./target-app/

# Step 2: Feed findings to LLM for triage
import json
import openai

with open("findings.json") as f:
    semgrep_results = json.load(f)

TRIAGE_PROMPT = """You are a security expert triaging Semgrep findings.
For each finding, determine:
1. Is this a TRUE positive or FALSE positive? Explain why.
2. If true positive: severity, exploitability, and recommended fix.
3. If false positive: why the code is actually safe.

Be conservative — when in doubt, flag it for manual review."""

for finding in semgrep_results.get("results", []):
    context = f"""
Rule: {finding['check_id']}
Severity: {finding['extra']['severity']}
Message: {finding['extra']['message']}
File: {finding['path']}:{finding['start']['line']}
Code: {finding['extra']['lines']}
"""
    response = openai.chat.completions.create(
        model="gpt-5",
        messages=[
            {"role": "system", "content": TRIAGE_PROMPT},
            {"role": "user", "content": context}
        ],
        temperature=0.1
    )
    print(f"\n--- {finding['path']}:{finding['start']['line']} ---")
    print(response.choices[0].message.content)

"""Deterministic checker for the pinned review exercise. Stdlib only.

Usage: python check-review.py review-fixture.py review-ground-truth.json reviewer-output.json
Exit 0 when every ground-truth finding is reproduced exactly (file, line, CWE,
severity, verification_status) and the flagged line really is the sink.
Exit 1 otherwise — a mismatch is a wrong review, not a checker bug.
"""
import json
import sys
from pathlib import Path


def fail(message):
    print(f"FAIL: {message}")
    raise SystemExit(1)


def main():
    if len(sys.argv) != 4:
        fail("usage: python check-review.py review-fixture.py review-ground-truth.json reviewer-output.json")
    fixture_lines = Path(sys.argv[1]).read_text(encoding="utf-8").splitlines()
    expected = json.loads(Path(sys.argv[2]).read_text(encoding="utf-8"))["findings"]
    actual_doc = json.loads(Path(sys.argv[3]).read_text(encoding="utf-8"))
    if isinstance(actual_doc, dict) and "findings" in actual_doc:
        actual = actual_doc["findings"]
    elif isinstance(actual_doc, dict):
        actual = [actual_doc]
    else:
        actual = actual_doc

    if len(actual) != len(expected):
        fail(f"expected {len(expected)} finding(s), got {len(actual)}")

    for want in expected:
        match = [got for got in actual if got.get("file") == want["file"] and got.get("line") == want["line"]]
        if not match:
            fail(f"missing finding at {want['file']}:{want['line']}")
        got = match[0]
        for field in ("file", "line", "cwe", "severity", "verification_status"):
            if got.get(field) != want[field]:
                fail(f"{want['file']}:{want['line']} field {field!r}: got {got.get(field)!r}, want {want[field]!r}")
        try:
            source_line = fixture_lines[want["line"] - 1]
        except IndexError:
            fail(f"line {want['line']} does not exist in {want['file']}")
        if "os.system" not in source_line:
            fail(f"{want['file']}:{want['line']} is not the sink (no os.system on that line)")

    print(f"PASS: {len(expected)} finding(s) reproduce file, line, CWE, severity and verification_status")
    raise SystemExit(0)


if __name__ == "__main__":
    main()