diff --git a/scripts/create_superuser.py b/scripts/create_superuser.py new file mode 100644 index 0000000..96288ff --- /dev/null +++ b/scripts/create_superuser.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""创建超级管理员（部署期本地执行，不经 HTTP 暴露）。 + +替代已删除的 /rbac/add_superuser.dspy —— 那个端点被授权给 any 角色且无任何鉴权， +任何人裸调 URL 即可创建 superuser/111111 超管并接管平台（安全事故，已移除）。 +超管初始化属于一次性引导动作，只应在部署期由运维在服务器本地执行。 + +用法： + py3/bin/python scripts/create_superuser.py --username superuser --password '强密码' + py3/bin/python scripts/create_superuser.py -u admin2 -p 'xxx' --orgid 0 + +安全约定： + - 拒绝弱密码（长度 < 10，或在常见弱口令表里） + - 已存在同名用户则拒绝（不静默改密码，避免误覆盖） + - 只能本地跑（依赖本机 conf/config.json 的 DB 配置） +""" + +import argparse +import asyncio +import getpass +import os +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', 'pkgs/rbac'): + sys.path.insert(0, os.path.join(APP_DIR, _p)) + +WEAK = { + '111111', '123456', '12345678', '123456789', 'password', 'admin', 'admin123', + 'passw0rd', 'qwerty', 'abc123', '1qaz2wsx', 'aa123456', 'Aa123456', +} + +MIN_LEN = 10 + + +def _check_password(pwd): + if not pwd: + return "密码不能为空" + if pwd in WEAK or pwd.lower() in WEAK: + return "密码在常见弱口令表中，拒绝使用" + if len(pwd) < MIN_LEN: + return f"密码长度须 >= {MIN_LEN}（当前 {len(pwd)}）" + kinds = sum([ + any(c.islower() for c in pwd), + any(c.isupper() for c in pwd), + any(c.isdigit() for c in pwd), + any(not c.isalnum() for c in pwd), + ]) + if kinds < 3: + return "密码须至少包含大写/小写/数字/符号中的 3 类" + return "" + + +async def main(args): + from appPublic.jsonConfig import getConfig + from appPublic.uniqueID import getID + from appPublic.dictObject import DictObject + from sqlor.dbpools import DBPools + from ahserver.globalEnv import password_encode + from rbac.check_perm import create_user + + config = getConfig(APP_DIR, {'workdir': APP_DIR}) + if not config.databases: + print("ERROR: conf/config.json 里没有 databases 配置", file=sys.stderr) + return 2 + DBPools(config.databases) + + dbname = 'pipeline' + async with DBPools().sqlorContext(dbname) as sor: + exist = await sor.sqlExe( + "SELECT id FROM users WHERE username=${u}$", {"u": args.username}) + await sor.sqlExe("COMMIT", {}) + if exist: + print(f"ERROR: 用户 {args.username} 已存在（拒绝覆盖，如需改密码请走 rbac 重置密码流程）", + file=sys.stderr) + return 3 + + ns = DictObject(**{ + "id": getID(), + "username": args.username, + "password": password_encode(args.password), + "orgid": args.orgid, + }) + roles = [{"orgtypeid": "owner", "roles": ["superuser", "admin"]}] + await create_user(sor, ns, roles) + await sor.sqlExe("COMMIT", {}) + + print(f"OK: 超管 {args.username} 已创建（orgid={args.orgid}, 角色 owner.superuser+owner.admin）") + print("提示：请立即登录验证，并妥善保管密码；本脚本不回显密码。") + return 0 + + +if __name__ == '__main__': + ap = argparse.ArgumentParser(description="创建超级管理员（部署期本地执行）") + ap.add_argument('-u', '--username', required=True, help="超管用户名") + ap.add_argument('-p', '--password', default=None, + help="超管密码（省略则交互式输入，不回显）") + ap.add_argument('--orgid', default='0', help="所属机构 id，默认 0（平台业主）") + a = ap.parse_args() + + if not a.password: + a.password = getpass.getpass("请输入超管密码（不回显）: ") + again = getpass.getpass("再次输入确认: ") + if a.password != again: + print("ERROR: 两次输入不一致", file=sys.stderr) + sys.exit(1) + + err = _check_password(a.password) + if err: + print(f"ERROR: {err}", file=sys.stderr) + sys.exit(1) + + sys.exit(asyncio.run(main(a)) or 0) diff --git a/scripts/security_check.py b/scripts/security_check.py new file mode 100644 index 0000000..77d0e7e --- /dev/null +++ b/scripts/security_check.py @@ -0,0 +1,185 @@ +#!/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)