Arm 开源 Metis：用 RAG + Agent 把漏洞挖掘的命中率拉到新水位

# minivuln_audit.py — 最小化 RAG Agent 漏洞审计原型
# 运行前安装依赖：pip install langchain chromadb openai tiktoken tree-sitter tree-sitter-python

import os
import json
from pathlib import Path
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# ── 1. 项目源码切片 & 入库 ──────────────────────────────

def ingest_project(project_dir: str, persist_dir: str = "./chroma_db"):
    """把项目源码按函数粒度切片，存入 Chroma 向量库"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,       # 控制每次检索的上下文大小
        chunk_overlap=100,
        separators=["\ndef ", "\nclass ", "\nif ", "\n"],
    )
    docs = []
    for path in Path(project_dir).rglob("*.py"):  # 可扩展 c/h 等后缀
        content = path.read_text(encoding="utf-8", errors="ignore")
        # 附带文件路径作为元数据，方便溯源
        chunks = splitter.split_text(content)
        for chunk in chunks:
            docs.append(Document(page_content=chunk, metadata={"source": str(path)}))

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectordb = Chroma.from_documents(docs, embeddings, persist_directory=persist_dir)
    vectordb.persist()
    print(f"已入库 {len(docs)} 个代码片段 → {persist_dir}")
    return vectordb

# ── 2. Agent 推理链定义 ──────────────────────────────────

ENTRY_PROMPT = """你是一个安全审计 Agent。以下是项目的代码片段：

{context}

请识别所有可能接收外部输入的"入口函数"（如网络请求处理、IPC、文件解析入口）。
对每个入口，列出：函数名、文件路径、输入来源类型。只输出 JSON 数组。"""

PATH_PROMPT = """基于已识别的入口函数：

{entries}

以下是相关代码上下文：

{context}

请从每个入口追踪数据流到危险 sink（memcpy/alloc/format/print 等无边界检查的调用）。
对每条路径，输出：入口→中间节点→sink，以及路径上是否有边界检查。JSON 数组格式。"""

VERIFY_PROMPT = """以下是候选漏洞路径：

{paths}

相关上下文：

{context}

逐条验证：路径是否真的可达？缺少的约束是否可被外部输入绕过？影响范围？
输出最终确认的漏洞列表，每条包含：路径、可利用性(高/中/低)、简要推理。JSON 格式。"""

# ── 3. 运行 Agent 链 ────────────────────────────────────

def run_audit(vectordb, project_name: str = "my_project"):
    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    # Step 1: 入口识别
    # 检索项目顶层模块和 handler 相关代码
    entry_docs = vectordb.similarity_search("handler request parse entry point", k=15)
    entry_context = "\n---\n".join(d.page_content for d in entry_docs)
    entries_raw = llm.invoke(ENTRY_PROMPT.format(context=entry_context)).content
    entries = json.loads(entries_raw)
    print(f"▸ 识别到 {len(entries)} 个入口函数")

    # Step 2: 路径追踪 — 对每个入口检索其调用链上下文
    all_paths = []
    for entry in entries:
        query = f"{entry['function_name']} call flow sink"
        path_docs = vectordb.similarity_search(query, k=10)
        path_context = "\n---\n".join(d.page_content for d in path_docs)
        paths_raw = llm.invoke(PATH_PROMPT.format(
            entries=json.dumps(entries, ensure_ascii=False),
            context=path_context
        )).content
        paths = json.loads(paths_raw)
        all_paths.extend(paths)
    print(f"▸ 发现 {len(all_paths)} 条候选路径")

    # Step 3: 可利用性验证
    verify_docs = vectordb.similarity_search("boundary check sanitizer validation", k=10)
    verify_context = "\n---\n".join(d.page_content for d in verify_docs)
    vulns_raw = llm.invoke(VERIFY_PROMPT.format(
        paths=json.dumps(all_paths, ensure_ascii=False),
        context=verify_context
    )).content
    vulns = json.loads(vulns_raw)
    print(f"▸ 最终确认 {len(vulns)} 个可利用漏洞")
    return vulns

# ── 4. 主流程 ────────────────────────────────────────────

if __name__ == "__main__":
    PROJECT_DIR = "./target_project"   # ← 替换为你要审计的项目路径
    os.environ.setdefault("OPENAI_API_KEY", "sk-xxx")  # ← 替换为你的 key

    vectordb = ingest_project(PROJECT_DIR)
    results = run_audit(vectordb)

    for v in results:
        print(f"\n🚨 {v.get('exploitability', '?')} | {v.get('path', '?')}")
        print(f"   推理: {v.get('reasoning', '?')}")