#!/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 代理（兼容保留，委托统一推理引擎） '/pipeline-llm/api/v1/', # token 鉴权的统一模型推理端点（模型治理模块） # 纯静态资源目录（图片/前端产物），匿名可读是设计如此 '/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', ) def _is_whitelisted(path): return any(path.startswith(p) for p in WHITELIST_PREFIXES) def _resolve_file(url_path): """URL 路径 → wwwroot 下的实际文件（含无扩展名回退）。找不到返回 None。 注意：目录型路径（CRUD 目录如 /discount/discount_list、静态目录如 /bricks） 返回 None 表示「不是单文件」，判断「路径是否存在」必须用 _path_exists()， 不能用本函数的 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 return None def _path_exists(url_path): """URL 路径在 wwwroot 下是否真实存在（文件 或 目录，跟随软链）。""" rel = url_path.lstrip('/') if not rel: return True # 根路径 if '*' in rel or rel.endswith('%'): return True # 通配授权另有 B 类检查 cand = os.path.join(WWWROOT, rel) if os.path.exists(cand): # os.path.exists 跟随软链，且覆盖目录 return True for ext in ('.dspy', '.ui', '.xterm', '.html'): if os.path.exists(cand + ext): return True return False 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': [], 'E': []} 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 [])}) # [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: 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] 指向不存在的路径（脏数据）。注意用 _path_exists 判存在性—— # 目录型路径（CRUD 目录、静态目录）不是单文件但真实存在，不算脏数据。 if not _path_exists(path): 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': '终端类端点存在匿名回退（未登录被当成某个用户）', 'E': '定时任务端点缺 client_ip=localhost 校验（白名单放行但无本机限制）', } total = sum(len(v) for v in findings.values()) print("=" * 60) print("部署期安全检查：any 授权 + 鉴权组合") print("=" * 60) for k in ('A', 'B', 'C', 'D', 'E'): items = findings[k] mark = '🔴' if k in ('A', 'B', 'D', 'E') 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']) + len(findings['E']) if blocking and not warn_only: print(f"\n❌ 存在 {blocking} 项阻断级问题（A/B/D/E），部署应中止。" 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)