yumoqing
5f18745ace
- scripts/create_superuser.py: 替代已删的 /rbac/add_superuser.dspy 后门， 部署期本地执行，强制强密码(>=10位/3类字符/拒弱口令)，已存在同名用户则拒绝 - scripts/security_check.py: 部署期扫「any 授权 + 无鉴权 + 危险动作」组合防回归， 另检 any 通配授权、指向已删文件的脏授权、终端端点匿名回退；A/B/D 类阻断部署
115 lines
4.2 KiB
Python
115 lines
4.2 KiB
Python
#!/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)
|