This commit is contained in:
QihuAI Design committed
1 parent
71fc502a92
commit
669c16b0a0
11 files changed
+4881
-4282
No files matched your search
@@ -0,0 +1,171 @@
|
||||
"""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(),
|
||||
}
|
||||
Reference in new issue
Block a user