用 MCP + Agentic AI 在代码审查中自动对齐安全需求与实现

// threat_models/api_user_query.json
{
  "component": "UserQueryAPI",
  "threats": [
    {
      "id": "TM-001",
      "description": "SQL injection via unsanitized user input",
      "mitigation": "All user inputs must be validated for type and length; use parameterized queries only",
      "severity": "high"
    },
    {
      "id": "TM-002",
      "description": "PII leakage in logs",
      "mitigation": "Logs must not contain email, phone, or real name; use anonymized identifiers",
      "severity": "medium"
    }
  ],
  "security_requirements": [
    "REQ-SEC-01: Parameterized queries for all DB operations",
    "REQ-SEC-02: PII must be redacted before logging"
  ]
}

# mcp_threat_server.py
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("threat-model-server")

THREAT_DIR = Path("./threat_models")

@mcp.tool()
def get_threat_model(component: str) -> str:
    """Retrieve the threat model for a given component name.
    Returns the full threat model JSON including threats and security requirements."""
    file_path = THREAT_DIR / f"{component.lower()}.json"
    if not file_path.exists():
        return f"No threat model found for component: {component}"
    return json.dumps(json.loads(file_path.read_text()), indent=2)

@mcp.tool()
def list_components() -> str:
    """List all components that have threat models available."""
    models = [f.stem for f in THREAT_DIR.glob("*.json")]
    return json.dumps(models)

@mcp.tool()
def check_requirement_coverage(component: str, requirement_id: str, pr_diff: str) -> str:
    """Check if a specific security requirement appears to be addressed in the PR diff.
    Returns an analysis of whether the implementation covers the requirement."""
    file_path = THREAT_DIR / f"{component.lower()}.json"
    if not file_path.exists():
        return f"No threat model for {component}"
    model = json.loads(file_path.read_text())
    reqs = model.get("security_requirements", [])
    matched = [r for r in reqs if requirement_id in r]
    if not matched:
        return f"Requirement {requirement_id} not found in threat model for {component}"
    return json.dumps({
        "requirement": matched[0],
        "pr_diff_summary": pr_diff[:500],
        "note": "Agent should analyze whether the diff implements this requirement"
    })

if __name__ == "__main__":
    mcp.run()

# 安装 MCP SDK
pip install mcp

# 启动 Server（默认 stdio 传输）
python mcp_threat_server.py

# review_agent.py
import subprocess
import json

def get_pr_diff(pr_number: int) -> str:
    """模拟获取 PR diff——实际中可调用 GitHub API"""
    # 示例：用 gh CLI 获取 diff
    result = subprocess.run(
        ["gh", "pr", "diff", str(pr_number)],
        capture_output=True, text=True
    )
    return result.stdout

def extract_component_from_diff(diff: str) -> str:
    """从 diff 中推断涉及的组件名——简化版"""
    # 实际中可用更智能的路径映射或 LLM 分类
    if "api/user_query" in diff or "user_query" in diff:
        return "UserQueryAPI"
    return "Unknown"

def run_security_gap_check(pr_number: int):
    diff = get_pr_diff(pr_number)
    component = extract_component_from_diff(diff)

    # 通过 MCP Client 查询威胁模型（这里用直接调用模拟）
    # 实际部署中 Agent 通过 MCP 协议与 Server 交互
    threat_model_path = f"./threat_models/{component.lower()}.json"
    try:
        model = json.loads(Path(threat_model_path).read_text())
    except FileNotFoundError:
        print(f"[WARN] No threat model for {component}, skipping security check")
        return

    print(f"=== Security Gap Report for PR #{pr_number} ===")
    print(f"Component: {component}")
    print(f"Threats in model: {len(model['threats'])}")
    print(f"Security requirements: {len(model['security_requirements'])}")
    print()

    # 检查每条安全需求是否在 diff 中有对应实现
    for req in model["security_requirements"]:
        req_keywords = extract_keywords(req)
        found = any(kw in diff.lower() for kw in req_keywords)
        status = "COVERED" if found else "GAP"
        print(f"  [{status}] {req}")
        if not found:
            print(f"         → No evidence in PR diff that this requirement is addressed")

    # 检查威胁是否被缓解
    for threat in model["threats"]:
        mitigation_keywords = extract_keywords(threat["mitigation"])
        found = any(kw in diff.lower() for kw in mitigation_keywords)
        status = "MITIGATED" if found else "UNMITIGATED"
        print(f"  [{status}] {threat['id']}: {threat['description']}")

def extract_keywords(text: str) -> list[str]:
    """从需求/缓解措施中提取关键检测词——简化版"""
    # 实际中可用 LLM 提取更精准的语义关键词
    keywords = []
    for word in text.lower().split():
        if len(word) > 4 and word not in {"must", "should", "all", "only", "before"}:
            keywords.append(word)
    return keywords

# 使用示例
run_security_gap_check(pr_number=42)

=== Security Gap Report for PR #42 ===
Component: UserQueryAPI
Threats in model: 2
Security requirements: 2

  [GAP] REQ-SEC-01: Parameterized queries for all DB operations
         → No evidence in PR diff that this requirement is addressed
  [GAP] REQ-SEC-02: PII must be redacted before logging
         → No evidence in PR diff that this requirement is addressed
  [UNMITIGATED] TM-001: SQL injection via unsanitized user input
  [UNMITIGATED] TM-002: PII leakage in logs