security: security_check.py 挂进 build.sh + 新增 E 类检查
- build.sh 第 12 步跑 security_check.py --warn-only(不阻断部署,打印清单); CI 可直接调脚本本体做硬门禁(A/B/D/E 返回 1) - 新增 [E] 类:白名单放行的定时任务端点必须自带 client_ip=localhost 校验, 防止白名单本身变成漏洞 - 静态资源目录通配(/rag/imgs/ 等)与 token 鉴权代理加入白名单,消除永久噪音
This commit is contained in:
parent
71566bea56
commit
fb853ca350
12
build.sh
12
build.sh
@ -219,4 +219,16 @@ if [ -f "$cdir/pkgs/accounting/wwwroot/api/fetch_forex_rates.dspy" ]; then
|
|||||||
fi
|
fi
|
||||||
|
|
||||||
chmod +x "$cdir/start.sh" "$cdir/stop.sh" 2>/dev/null || true
|
chmod +x "$cdir/start.sh" "$cdir/stop.sh" 2>/dev/null || true
|
||||||
|
|
||||||
|
# 12. 部署期安全检查:扫「any 授权 + 端点无鉴权 + 危险动作」等危险组合(防回归)
|
||||||
|
# 不阻断部署(与本脚本其它步骤的 WARN 风格一致),但会打印清单;
|
||||||
|
# CI 里可直接调 scripts/security_check.py(A/B/D 类返回退出码 1)做硬门禁。
|
||||||
|
echo "=== Security check (RBAC any grants × endpoint auth) ==="
|
||||||
|
if [ -f "$cdir/scripts/security_check.py" ]; then
|
||||||
|
"$cdir/py3/bin/python" "$cdir/scripts/security_check.py" --warn-only \
|
||||||
|
|| echo " WARN: security_check.py failed (DB not ready?)"
|
||||||
|
else
|
||||||
|
echo " WARN: scripts/security_check.py not found"
|
||||||
|
fi
|
||||||
|
|
||||||
echo "=== Build complete ==="
|
echo "=== Build complete ==="
|
||||||
|
|||||||
@ -61,6 +61,18 @@ WHITELIST_PREFIXES = (
|
|||||||
'/rbac/user/reset_password',
|
'/rbac/user/reset_password',
|
||||||
'/i18n', '/bricks/', '/assets/', '/imgs/', '/download/',
|
'/i18n', '/bricks/', '/assets/', '/imgs/', '/download/',
|
||||||
'/pipeline_core/api/llm_v1/', # token 鉴权的 LLM 代理
|
'/pipeline_core/api/llm_v1/', # token 鉴权的 LLM 代理
|
||||||
|
# 纯静态资源目录(图片/前端产物),匿名可读是设计如此
|
||||||
|
'/rag/imgs/', '/reallife_asset/imgs/', '/univer-office/',
|
||||||
|
# 定时任务入口:不走登录态,靠 client_ip=localhost 隔离(应用端口不对外开放 +
|
||||||
|
# nginx $proxy_add_x_forwarded_for 追加模式使外部伪造不成立)
|
||||||
|
'/appbase/cron/switch_bizdate.dspy',
|
||||||
|
'/accounting/api/fetch_forex_rates.dspy',
|
||||||
|
)
|
||||||
|
|
||||||
|
# 定时任务类端点必须自带 localhost 校验,否则视为 A 类风险
|
||||||
|
LOCALHOST_GUARD_REQUIRED = (
|
||||||
|
'/appbase/cron/switch_bizdate.dspy',
|
||||||
|
'/accounting/api/fetch_forex_rates.dspy',
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@ -104,7 +116,7 @@ async def main(warn_only):
|
|||||||
return 0
|
return 0
|
||||||
DBPools(config.databases)
|
DBPools(config.databases)
|
||||||
|
|
||||||
findings = {'A': [], 'B': [], 'C': [], 'D': []}
|
findings = {'A': [], 'B': [], 'C': [], 'D': [], 'E': []}
|
||||||
|
|
||||||
async with DBPools().sqlorContext('pipeline') as sor:
|
async with DBPools().sqlorContext('pipeline') as sor:
|
||||||
recs = await sor.sqlExe(
|
recs = await sor.sqlExe(
|
||||||
@ -113,6 +125,18 @@ async def main(warn_only):
|
|||||||
await sor.sqlExe("COMMIT", {})
|
await sor.sqlExe("COMMIT", {})
|
||||||
any_paths = sorted({(getattr(r, 'path', '') or '').strip() for r in (recs or [])})
|
any_paths = sorted({(getattr(r, 'path', '') or '').strip() for r in (recs or [])})
|
||||||
|
|
||||||
|
# [E] 白名单里的定时任务端点必须自带 localhost 校验 —— 否则白名单本身成为漏洞
|
||||||
|
for p in LOCALHOST_GUARD_REQUIRED:
|
||||||
|
fp = _resolve_file(p)
|
||||||
|
if fp is None:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
src = open(fp, encoding='utf-8', errors='ignore').read()
|
||||||
|
except OSError:
|
||||||
|
continue
|
||||||
|
if 'client_ip' not in src or '127.0.0.1' not in src:
|
||||||
|
findings['E'].append(f"{p} (定时任务端点缺 client_ip=localhost 校验)")
|
||||||
|
|
||||||
for path in any_paths:
|
for path in any_paths:
|
||||||
if not path or _is_whitelisted(path):
|
if not path or _is_whitelisted(path):
|
||||||
continue
|
continue
|
||||||
@ -152,22 +176,23 @@ async def main(warn_only):
|
|||||||
'B': 'any 通配授权(覆盖整个模块目录,安全全靠各端点自觉)',
|
'B': 'any 通配授权(覆盖整个模块目录,安全全靠各端点自觉)',
|
||||||
'C': 'any 授权指向不存在的文件(permission 表脏数据,应清理)',
|
'C': 'any 授权指向不存在的文件(permission 表脏数据,应清理)',
|
||||||
'D': '终端类端点存在匿名回退(未登录被当成某个用户)',
|
'D': '终端类端点存在匿名回退(未登录被当成某个用户)',
|
||||||
|
'E': '定时任务端点缺 client_ip=localhost 校验(白名单放行但无本机限制)',
|
||||||
}
|
}
|
||||||
total = sum(len(v) for v in findings.values())
|
total = sum(len(v) for v in findings.values())
|
||||||
print("=" * 60)
|
print("=" * 60)
|
||||||
print("部署期安全检查:any 授权 + 鉴权组合")
|
print("部署期安全检查:any 授权 + 鉴权组合")
|
||||||
print("=" * 60)
|
print("=" * 60)
|
||||||
for k in ('A', 'B', 'C', 'D'):
|
for k in ('A', 'B', 'C', 'D', 'E'):
|
||||||
items = findings[k]
|
items = findings[k]
|
||||||
mark = '🔴' if k in ('A', 'B', 'D') else '🟠'
|
mark = '🔴' if k in ('A', 'B', 'D', 'E') else '🟠'
|
||||||
print(f"\n[{k}] {titles[k]}: {len(items)} 项")
|
print(f"\n[{k}] {titles[k]}: {len(items)} 项")
|
||||||
for it in items:
|
for it in items:
|
||||||
print(f" {mark} {it}")
|
print(f" {mark} {it}")
|
||||||
print(f"\n合计 {total} 项")
|
print(f"\n合计 {total} 项")
|
||||||
|
|
||||||
blocking = len(findings['A']) + len(findings['B']) + len(findings['D'])
|
blocking = len(findings['A']) + len(findings['B']) + len(findings['D']) + len(findings['E'])
|
||||||
if blocking and not warn_only:
|
if blocking and not warn_only:
|
||||||
print(f"\n❌ 存在 {blocking} 项阻断级问题(A/B/D),部署应中止。"
|
print(f"\n❌ 存在 {blocking} 项阻断级问题(A/B/D/E),部署应中止。"
|
||||||
f"\n 修法:删除多余 any 授权(改 logined 或具体角色)+ 端点内补鉴权;"
|
f"\n 修法:删除多余 any 授权(改 logined 或具体角色)+ 端点内补鉴权;"
|
||||||
f"\n 改完执行 redis-cli DEL sc:rbac:role_perms 并重启。")
|
f"\n 改完执行 redis-cli DEL sc:rbac:role_perms 并重启。")
|
||||||
return 1
|
return 1
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user