MDASH：微软用上百个 AI Agent 组团挖漏洞

"""
mini_mdash.py — 简化版多 Agent 代码审计流水线
依赖: pip install openai
运行前设置: export OPENAI_API_KEY=sk-xxx
用法: python mini_mdash.py <代码文件路径>
"""

import os
import sys
import json
from openai import OpenAI

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

AGENT_PROMPTS = {
    "scanner": (
        "你是代码安全扫描 Agent。阅读以下代码，列出所有潜在安全风险点。"
        "每条风险点用一行输出，格式: [行号/区域] 风险类型 - 简要描述。"
        "不要做深度验证，只做快速标记。"
    ),
    "validator": (
        "你是漏洞验证 Agent。对以下风险点列表逐条分析："
        "判断每个风险点是否可被外部输入触发（路径可达性）。"
        "输出 JSON 数组，每个元素包含: risk(原风险描述), reachable(bool), reason(可达/不可达的理由)。"
    ),
    "debater_pro": (
        "你是辩论正方 Agent。针对以下验证结果中标记为 reachable 的风险点，"
        "为每个点补充攻击路径细节：输入如何到达、约束如何绕过。"
        "输出 JSON 数组，每个元素: risk, attack_path, confidence(0-1)。"
    ),
    "debater_con": (
        "你是辩论反方 Agent。针对以下验证结果中标记为 reachable 的风险点，"
        "为每个点提出反驳：为什么实际不可利用（防御机制、运行时约束等）。"
        "输出 JSON 数组，每个元素: risk, rebuttal, confidence(0-1)。"
    ),
    "judge": (
        "你是裁决 Agent。综合正方攻击路径和反方反驳，对每个风险点做出最终判定。"
        "输出 JSON 数组，每个元素: risk, verdict(vulnerable/safe/uncertain), "
        "reason(综合理由), suggested_action(建议修复或进一步验证)。"
    ),
}

def call_agent(role: str, context: str) -> str:
    prompt = AGENT_PROMPTS[role] + "\n\n" + context
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content

def run_pipeline(code: str):
    # Step 1: 扫描
    scan_result = call_agent("scanner", code)
    print("=== 扫描结果 ===")
    print(scan_result)

    # Step 2: 验证
    validation_result = call_agent("validator", scan_result)
    print("\n=== 验证结果 ===")
    print(validation_result)

    # Step 3: 辩论
    pro_result = call_agent("debater_pro", validation_result)
    con_result = call_agent("debater_con", validation_result)
    print("\n=== 正方论证 ===")
    print(pro_result)
    print("\n=== 反方反驳 ===")
    print(con_result)

    # Step 4: 裁决
    debate_context = f"正方论证:\n{pro_result}\n\n反方反驳:\n{con_result}"
    final_result = call_agent("judge", debate_context)
    print("\n=== 最终裁决 ===")
    print(final_result)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python mini_mdash.py <代码文件路径>")
        sys.exit(1)
    with open(sys.argv[1], "r", encoding="utf-8") as f:
        code_content = f.read()
    run_pipeline(code_content)

# 准备一个测试文件
cat > test_vuln.c << 'EOF'
#include <stdio.h>
#include <string.h>

void process_input(char *user_data) {
    char buffer[64];
    strcpy(buffer, user_data);  // 明显的缓冲区溢出
    printf("Processed: %s\n", buffer);
}

int main(int argc, char **argv) {
    if (argc > 1) {
        process_input(argv[1]);
    }
    return 0;
}
EOF

# 运行审计流水线
export OPENAI_API_KEY=sk-your-key-here
python mini_mdash.py test_vuln.c