yumoqing
5f18745ace
- scripts/create_superuser.py: 替代已删的 /rbac/add_superuser.dspy 后门， 部署期本地执行，强制强密码(>=10位/3类字符/拒弱口令)，已存在同名用户则拒绝 - scripts/security_check.py: 部署期扫「any 授权 + 无鉴权 + 危险动作」组合防回归， 另检 any 通配授权、指向已删文件的脏授权、终端端点匿名回退；A/B/D 类阻断部署
186 lines
7.4 KiB
Python
186 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""部署期安全检查：扫「any 授权 + 无鉴权」的危险组合，防回归。
|
||
|
||
背景：2026-08-25 安全审计实锤两个未授权接管路径——
|
||
1. /rbac/add_superuser.dspy 授权给 any 且文件内零鉴权 → 未登录创建超管 → 接管平台
|
||
2. /pipeline-sdlc/% 通配授权给 any（rbac 的 % 是前缀通配）→ 整模块对未登录开放，
|
||
实测未登录可读 sd_deploy_envs（部署主机/SSH 账号/路径全泄露）
|
||
|
||
这类洞的成因是「RBAC 授权」与「端点内部鉴权」两层各自都可能漏，
|
||
组合起来才安全。本脚本在部署期把危险组合拦下来。
|
||
|
||
用法：
|
||
py3/bin/python scripts/security_check.py # 检查，有问题退出码 1
|
||
py3/bin/python scripts/security_check.py --warn-only # 只告警，退出码恒 0
|
||
|
||
检查项：
|
||
[A] any 授权指向的端点文件里没有鉴权调用（get_user / check_*_owner / verify_*_token）
|
||
[B] any 授权用了通配符（** 或 %）覆盖整个模块目录
|
||
[C] any 授权指向已不存在的文件（permission 表脏数据，应清理）
|
||
[D] .xterm 终端类端点存在匿名回退（uid = 'user-01' 之类）
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
APP_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, APP_DIR)
|
||
for _p in ('pkgs/apppublic', 'pkgs/sqlor', 'pkgs/ahserver'):
|
||
sys.path.insert(0, os.path.join(APP_DIR, _p))
|
||
|
||
WWWROOT = os.path.join(APP_DIR, 'wwwroot')
|
||
|
||
# 视为「已鉴权」的调用特征
|
||
AUTH_PATTERNS = [
|
||
r'get_user\(\)',
|
||
r'check_project_owner', r'check_task_owner', r'check_tenant_owner',
|
||
r'verify_llm_token', r'proxy_chat_completion', # token 鉴权端点
|
||
r'check_wechat_signature', r'verify_signature', # 回调签名校验
|
||
]
|
||
|
||
# 危险动作特征（写/删/执行）——只读端点漏鉴权风险较低，单独归类
|
||
DANGER_PATTERNS = [
|
||
(r'\bsor\.C\(|insert\s+into', '写库'),
|
||
(r'\bsor\.U\(|\bupdate\s+\w+\s+set', '改库'),
|
||
(r'\bsor\.D\(|delete\s+from', '删库'),
|
||
(r'run_shell|subprocess|create_subprocess|os\.system', '执行命令'),
|
||
(r'cmdargs', '终端进程'),
|
||
(r'create_user|password_encode', '账号操作'),
|
||
(r'shutil\.(copy|move|rmtree)|os\.remove|open\([^)]*[\'"]w', '写文件'),
|
||
]
|
||
|
||
# 白名单：设计上必须匿名可访问且已自带鉴权/无危险动作的端点
|
||
WHITELIST_PREFIXES = (
|
||
'/rbac/user/up_login.dspy', '/rbac/user/login', '/rbac/user/register',
|
||
'/rbac/userpassword_login', '/rbac/user/code_login.dspy',
|
||
'/rbac/user/sms_register.dspy', '/rbac/gen_sms_code.dspy',
|
||
'/rbac/phone_login.dspy', '/rbac/user/logout.dspy',
|
||
'/rbac/user/reset_password',
|
||
'/i18n', '/bricks/', '/assets/', '/imgs/', '/download/',
|
||
'/pipeline_core/api/llm_v1/', # token 鉴权的 LLM 代理
|
||
)
|
||
|
||
|
||
def _is_whitelisted(path):
|
||
return any(path.startswith(p) for p in WHITELIST_PREFIXES)
|
||
|
||
|
||
def _resolve_file(url_path):
|
||
"""URL 路径 → wwwroot 下的实际文件（含无扩展名回退）。找不到返回 None。"""
|
||
rel = url_path.lstrip('/')
|
||
if not rel or '*' in rel or rel.endswith('%'):
|
||
return None
|
||
cand = os.path.join(WWWROOT, rel)
|
||
if os.path.isfile(cand):
|
||
return cand
|
||
for ext in ('.dspy', '.ui', '.xterm', '.html'):
|
||
if os.path.isfile(cand + ext):
|
||
return cand + ext
|
||
if os.path.isdir(cand):
|
||
return None
|
||
return None
|
||
|
||
|
||
def _scan_file(fp):
|
||
try:
|
||
src = open(fp, encoding='utf-8', errors='ignore').read()
|
||
except OSError:
|
||
return True, []
|
||
has_auth = any(re.search(p, src) for p in AUTH_PATTERNS)
|
||
dangers = [label for pat, label in DANGER_PATTERNS if re.search(pat, src, re.I)]
|
||
return has_auth, dangers
|
||
|
||
|
||
async def main(warn_only):
|
||
from appPublic.jsonConfig import getConfig
|
||
from sqlor.dbpools import DBPools
|
||
|
||
config = getConfig(APP_DIR, {'workdir': APP_DIR})
|
||
if not config.databases:
|
||
print("SKIP: 无 databases 配置，跳过 RBAC 检查")
|
||
return 0
|
||
DBPools(config.databases)
|
||
|
||
findings = {'A': [], 'B': [], 'C': [], 'D': []}
|
||
|
||
async with DBPools().sqlorContext('pipeline') as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT p.path FROM permission p JOIN rolepermission rp ON rp.permid=p.id "
|
||
"WHERE rp.roleid='any'", {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
any_paths = sorted({(getattr(r, 'path', '') or '').strip() for r in (recs or [])})
|
||
|
||
for path in any_paths:
|
||
if not path or _is_whitelisted(path):
|
||
continue
|
||
# [B] 通配授权
|
||
if path.endswith('**') or path.endswith('%') or path.endswith('/*'):
|
||
findings['B'].append(path)
|
||
continue
|
||
fp = _resolve_file(path)
|
||
if fp is None:
|
||
# [C] 指向不存在的文件（脏数据）；静态资源目录/根路径跳过
|
||
if path not in ('/', '') and not path.endswith(('.js', '.css', '.png', '.svg',
|
||
'.ico', '.woff', '.woff2', '.ttf')):
|
||
findings['C'].append(path)
|
||
continue
|
||
if fp.endswith(('.js', '.css', '.png', '.svg', '.ico', '.md')):
|
||
continue
|
||
has_auth, dangers = _scan_file(fp)
|
||
# [A] any 授权 + 无鉴权 + 有危险动作
|
||
if not has_auth and dangers:
|
||
findings['A'].append(f"{path} [{'+'.join(dangers)}]")
|
||
|
||
# [D] 终端类端点匿名回退
|
||
for root, _dirs, files in os.walk(WWWROOT):
|
||
for fn in files:
|
||
if not fn.endswith('.xterm'):
|
||
continue
|
||
fp = os.path.join(root, fn)
|
||
try:
|
||
src = open(fp, encoding='utf-8', errors='ignore').read()
|
||
except OSError:
|
||
continue
|
||
if re.search(r"if\s+not\s+uid\s*:\s*\n\s+uid\s*=\s*['\"]", src):
|
||
findings['D'].append(os.path.relpath(fp, APP_DIR))
|
||
|
||
titles = {
|
||
'A': 'any 授权 + 端点无鉴权 + 含危险动作（未授权写/删/执行）',
|
||
'B': 'any 通配授权（覆盖整个模块目录，安全全靠各端点自觉）',
|
||
'C': 'any 授权指向不存在的文件（permission 表脏数据，应清理）',
|
||
'D': '终端类端点存在匿名回退（未登录被当成某个用户）',
|
||
}
|
||
total = sum(len(v) for v in findings.values())
|
||
print("=" * 60)
|
||
print("部署期安全检查：any 授权 + 鉴权组合")
|
||
print("=" * 60)
|
||
for k in ('A', 'B', 'C', 'D'):
|
||
items = findings[k]
|
||
mark = '🔴' if k in ('A', 'B', 'D') else '🟠'
|
||
print(f"\n[{k}] {titles[k]}: {len(items)} 项")
|
||
for it in items:
|
||
print(f" {mark} {it}")
|
||
print(f"\n合计 {total} 项")
|
||
|
||
blocking = len(findings['A']) + len(findings['B']) + len(findings['D'])
|
||
if blocking and not warn_only:
|
||
print(f"\n❌ 存在 {blocking} 项阻断级问题（A/B/D），部署应中止。"
|
||
f"\n 修法：删除多余 any 授权（改 logined 或具体角色）+ 端点内补鉴权；"
|
||
f"\n 改完执行 redis-cli DEL sc:rbac:role_perms 并重启。")
|
||
return 1
|
||
if total:
|
||
print("\n⚠️ 存在非阻断项（C 类脏数据建议清理）")
|
||
else:
|
||
print("\n✅ 未发现危险组合")
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
ap = argparse.ArgumentParser(description="部署期 RBAC 安全检查")
|
||
ap.add_argument('--warn-only', action='store_true', help="只告警不阻断（退出码恒 0）")
|
||
a = ap.parse_args()
|
||
sys.exit(asyncio.run(main(a.warn_only)) or 0)
|