qihuaiflow/backend/python/app/core/app_security.py

"""Phase 4.1 应用市场安全扫描引擎。

scan_market_app(...) 对 skill_content / agent_endpoint / mcp 配置 / 依赖声明做
静态规则扫描，返回结构化结果：
    {
        "status": "pass" | "warn" | "fail",
        "score": 0-100,
        "findings": [{"level": "fail"|"warn"|"info", "rule": str, "detail": str}],
        "scanned_at": ISO 时间
    }
规则覆盖：危险 Python 调用、破坏性 shell、提示词注入、外传地址、硬编码密钥、
代码混淆、MCP 传输白名单、Agent 端点私网、依赖声明校验。
"""
import re
from datetime import datetime
from urllib.parse import urlparse

# 危险 Python 调用（risk 评估）
_DANGEROUS_PY = [
    ("os.system", "可执行任意 shell 命令"),
    ("subprocess", "可启动任意子进程"),
    ("__import__", "可动态加载任意模块"),
    ("eval(", "可执行任意代码"),
    ("exec(", "可执行任意代码"),
    ("pickle.loads", "反序列化可触发任意代码执行"),
    ("ctypes", "可调用底层 C 库"),
    ("marshal.loads", "反序列化风险"),
    ("compile(", "可动态构造代码"),
]

# 破坏性 shell 片段
_DESTRUCTIVE_SHELL = [
    ("rm -rf /", "全盘递归删除"),
    ("rm -rf ~", "删除用户主目录"),
    ("mkfs", "格式化文件系统"),
    ("dd if=", "直接写磁盘设备"),
    (":(){", "fork bomb 拒绝服务"),
    ("chmod -R 777 /", "篡改系统权限"),
    ("curl ", "下载远程脚本/数据"),
    ("wget ", "下载远程脚本/数据"),
]

# 提示词注入关键字（小写匹配）
_PROMPT_INJECTION = [
    "ignore previous instructions",
    "忽略之前的指令",
    "忽略以上",
    "disregard previous",
    "忽略所有",
    "你现在必须",
    "system prompt",
    "ignore all previous",
    "你现在是",
]

# 外泄目标（命中即高风险）
_EXFIL_HOSTS = [
    "pastebin.com", "pastebin", "webhook", "requestbin", "ngrok",
    "hookbin", "pipedream", "discord.com/api/webhooks", "telegram",
    "transfer.sh", "0x0.st",
]

# 硬编码密钥模式
_SECRET_PATTERNS = [
    (r"(api[_-]?key|secret|token|password)\s*[:=]\s*[\"']?[A-Za-z0-9_\-]{16,}[\"']?", "疑似硬编码密钥"),
    (r"sk-[A-Za-z0-9]{20,}", "疑似 OpenAI 风格密钥"),
    (r"AKIA[0-9A-Z]{16}", "疑似 AWS Access Key"),
]

# MCP 传输白名单
_MCP_TRANSPORT_ALLOW = {"stdio", "sse", "streamable_http"}

# 依赖白名单（允许声明的运行时依赖前缀）
_DEP_ALLOW = [
    "python", "node", "requests", "httpx", "openai", "langchain",
    "fastapi", "flask", "numpy", "pandas", "pillow", "mcp",
]


def _host_of(url):
    try:
        return (urlparse(url).hostname or "").lower()
    except Exception:
        return ""


def scan_market_app(app_type, skill_content=None, agent_endpoint=None,
                    mcp_transport=None, mcp_config=None, dependencies=None):
    findings = []

    content = skill_content or ""
    low = content.lower()

    # 1. 危险 Python
    for token, detail in _DANGEROUS_PY:
        if token in low:
            lvl = "fail" if token in ("eval(", "exec(", "os.system", "subprocess", "__import__") else "warn"
            findings.append({"level": lvl, "rule": "dangerous_python", "detail": "检测到 " + token + "：" + detail})

    # 2. 破坏性 shell
    for token, detail in _DESTRUCTIVE_SHELL:
        if token in low:
            findings.append({"level": "fail", "rule": "destructive_shell", "detail": "检测到 " + token + "：" + detail})

    # 3. 提示词注入（命中一个即记一条）
    for kw in _PROMPT_INJECTION:
        if kw in low:
            findings.append({"level": "warn", "rule": "prompt_injection", "detail": "疑似提示词注入关键字：" + kw})
            break

    # 4. 外泄 URL
    urls = re.findall(r"https?://[^\s\"'<>\)\]\}]+", content)
    for u in urls:
        h = _host_of(u)
        for bad in _EXFIL_HOSTS:
            if bad in h:
                findings.append({"level": "fail", "rule": "exfil_url", "detail": "检测到可疑外传地址：" + u})
                break

    # 5. 硬编码密钥（命中一个即记一条）
    for pat, detail in _SECRET_PATTERNS:
        if re.search(pat, content):
            findings.append({"level": "warn", "rule": "hardcoded_secret", "detail": detail})
            break

    # 6. 混淆 base64 + exec/eval
    if re.search(r"base64\.b64decode", content) and ("exec(" in low or "eval(" in low):
        findings.append({"level": "fail", "rule": "obfuscation", "detail": "检测到 base64 解码后执行，疑似代码混淆"})

    # 7. MCP 传输白名单
    if app_type == "mcp":
        t = (mcp_transport or "").lower()
        if t and t not in _MCP_TRANSPORT_ALLOW:
            findings.append({"level": "fail", "rule": "mcp_transport", "detail": "不支持的 MCP 传输类型：" + str(mcp_transport)})

    # 8. Agent 端点私网/回环检查
    if app_type == "agent" and agent_endpoint:
        h = _host_of(agent_endpoint)
        if h in ("localhost", "127.0.0.1", "0.0.0.0") or h.startswith("192.168.") or h.startswith("10.") or h.startswith("172."):
            findings.append({"level": "warn", "rule": "agent_endpoint_private", "detail": "Agent 端点指向内网/回环地址：" + agent_endpoint + "（平台可能无法访问）"})

    # 9. 依赖声明校验
    if dependencies:
        if not isinstance(dependencies, list):
            findings.append({"level": "warn", "rule": "dependencies_format", "detail": "依赖声明应为数组"})
        else:
            for dep in dependencies:
                name = (dep.get("name") if isinstance(dep, dict) else str(dep)).lower()
                if name and not any(name.startswith(a) for a in _DEP_ALLOW):
                    findings.append({"level": "info", "rule": "dependency_unknown", "detail": "未识别的依赖：" + name})

    # 判定
    fails = [f for f in findings if f["level"] == "fail"]
    warns = [f for f in findings if f["level"] == "warn"]
    if fails:
        status = "fail"
    elif warns:
        status = "warn"
    else:
        status = "pass"

    score = 100 - len(fails) * 25 - len(warns) * 8
    if score < 0:
        score = 0

    return {
        "status": status,
        "score": score,
        "findings": findings,
        "scanned_at": datetime.now().isoformat(),
    }