security: 新增脏授权清理脚本 + 修正目录型路径误判
clean_dirty_perms.py：清理指向不存在路径的 any 脏授权。默认预演(--dry-run)， --apply 才动库且先导出可回滚备份 SQL；保护名单(/rbac/** 登录流程、/bricks /imgs /assets /i18n 静态资源、根路径)永不删；通配路径跳过交人工决策。 实测清掉 216 条(llmage/rag/public/msp/supplychain 等未安装模块残留 + xls2ui 历史 bug 产生的 //i18n_getmsgs// 双斜杠畸形路径)，any 授权 801→585。 security_check.py 修正关键误判：原 _resolve_file 把目录型路径(CRUD 列表目录、 /bricks /imgs 静态目录)判成「文件不存在」→ 会把真实存在的目录误报成脏授权， 照该清单盲删会删掉登录页必需的静态资源授权。新增 _path_exists() (os.path.exists 跟随软链、目录也算存在)专门判存在性。
This commit is contained in:
parent
e756236272
commit
f1af8c882f
163
scripts/clean_dirty_perms.py
Normal file
163
scripts/clean_dirty_perms.py
Normal file
@ -0,0 +1,163 @@
|
||||
#!/usr/bin/env python3
|
||||
"""清理 permission/rolepermission 表里指向「已不存在路径」的 any 脏授权。
|
||||
|
||||
背景：pipeline 的 permission 表混入了从 sage 平台导入的历史授权（cpcc/dagflow/
|
||||
discount/llmage/dashboard_for_sage 等模块的路径），这些模块并未安装在本部署，
|
||||
授权行成了脏数据 —— 干扰安全审计（噪音掩盖真问题），也是「装上同名模块即意外
|
||||
放开匿名访问」的隐患。
|
||||
|
||||
安全约定（避免误删）：
|
||||
1. 只处理 roleid='any' 的授权行；其他角色一律不动。
|
||||
2. 「存在性」判定用 os.path.exists（跟随软链）+ 无扩展名回退（.dspy/.ui/.xterm/.html），
|
||||
且**目录也算存在** —— CRUD 目录(/discount/discount_list)、静态目录(/bricks、/imgs)
|
||||
不是单文件但真实存在，绝不能当脏数据删。
|
||||
3. 通配路径（含 * 或以 % 结尾）跳过 —— 那是 B 类问题，另行人工决策。
|
||||
4. 保护名单：登录/静态资源等匿名必需路径永不删。
|
||||
5. 先把待删行导出为可回滚 SQL 再删；--dry-run 只看不删（默认）。
|
||||
|
||||
用法：
|
||||
py3/bin/python scripts/clean_dirty_perms.py # 预演（默认，不改库）
|
||||
py3/bin/python scripts/clean_dirty_perms.py --apply # 实际删除（先写备份）
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
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')
|
||||
|
||||
# 永不删除（匿名必需：登录流程 + 静态资源 + 根路径）
|
||||
PROTECTED_PREFIXES = (
|
||||
'/', '/index.ui', '/index.html',
|
||||
'/rbac/', '/i18n', '/bricks', '/assets', '/imgs', '/favicon.ico',
|
||||
'/download', '/idfile',
|
||||
)
|
||||
|
||||
|
||||
def _path_exists(url_path):
|
||||
"""URL 路径在 wwwroot 下是否真实存在（文件或目录，跟随软链）。"""
|
||||
rel = url_path.lstrip('/')
|
||||
if not rel:
|
||||
return True
|
||||
cand = os.path.join(WWWROOT, rel)
|
||||
if os.path.exists(cand):
|
||||
return True
|
||||
for ext in ('.dspy', '.ui', '.xterm', '.html'):
|
||||
if os.path.exists(cand + ext):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_protected(path):
|
||||
if path in ('/', ''):
|
||||
return True
|
||||
# 登录流程整支保护
|
||||
if path.startswith('/rbac/'):
|
||||
return True
|
||||
for p in PROTECTED_PREFIXES:
|
||||
if p == '/':
|
||||
continue
|
||||
if path == p or path.startswith(p.rstrip('/') + '/'):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def main(apply_it):
|
||||
from appPublic.jsonConfig import getConfig
|
||||
from sqlor.dbpools import DBPools
|
||||
|
||||
config = getConfig(APP_DIR, {'workdir': APP_DIR})
|
||||
if not config.databases:
|
||||
print("ERROR: 无 databases 配置", file=sys.stderr)
|
||||
return 2
|
||||
DBPools(config.databases)
|
||||
|
||||
async with DBPools().sqlorContext('pipeline') as sor:
|
||||
rows = await sor.sqlExe(
|
||||
"SELECT rp.id AS rp_id, p.id AS perm_id, p.path "
|
||||
"FROM rolepermission rp JOIN permission p ON p.id=rp.permid "
|
||||
"WHERE rp.roleid='any' ORDER BY p.path", {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
total = len(rows or [])
|
||||
to_delete = []
|
||||
kept_wild = []
|
||||
kept_protected = []
|
||||
kept_exists = 0
|
||||
|
||||
for r in (rows or []):
|
||||
path = (getattr(r, 'path', '') or '').strip()
|
||||
rp_id = getattr(r, 'rp_id', '')
|
||||
perm_id = getattr(r, 'perm_id', '')
|
||||
if not path:
|
||||
continue
|
||||
if '*' in path or path.endswith('%'):
|
||||
kept_wild.append(path)
|
||||
continue
|
||||
if _is_protected(path):
|
||||
kept_protected.append(path)
|
||||
continue
|
||||
if _path_exists(path):
|
||||
kept_exists += 1
|
||||
continue
|
||||
to_delete.append((rp_id, perm_id, path))
|
||||
|
||||
print("=" * 64)
|
||||
print(f"any 授权总数 {total}")
|
||||
print(f" 路径真实存在，保留: {kept_exists}")
|
||||
print(f" 通配授权，跳过(B类另议): {len(kept_wild)}")
|
||||
print(f" 保护名单(登录/静态)，保留: {len(kept_protected)}")
|
||||
print(f" 路径不存在，待删: {len(to_delete)}")
|
||||
print("=" * 64)
|
||||
for _rp, _pm, path in to_delete:
|
||||
print(f" - {path}")
|
||||
|
||||
if not to_delete:
|
||||
print("\n无需清理")
|
||||
return 0
|
||||
|
||||
if not apply_it:
|
||||
print(f"\n[预演] 未改动数据库。确认无误后加 --apply 执行。")
|
||||
return 0
|
||||
|
||||
# 备份为可回滚 SQL
|
||||
ts = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
bak = os.path.join(APP_DIR, 'logs', f'dirty_perms_backup_{ts}.sql')
|
||||
os.makedirs(os.path.dirname(bak), exist_ok=True)
|
||||
with open(bak, 'w', encoding='utf-8') as f:
|
||||
f.write(f"-- any 脏授权清理备份 {ts}\n")
|
||||
f.write("-- 回滚：直接执行本文件即可恢复被删的 rolepermission 行\n")
|
||||
for rp_id, perm_id, path in to_delete:
|
||||
esc = path.replace("'", "''")
|
||||
f.write(
|
||||
f"INSERT INTO rolepermission (id, roleid, permid) "
|
||||
f"VALUES ('{rp_id}', 'any', '{perm_id}'); -- {esc}\n")
|
||||
print(f"\n备份已写入: {bak}")
|
||||
|
||||
deleted = 0
|
||||
for rp_id, _perm_id, _path in to_delete:
|
||||
await sor.sqlExe(
|
||||
"DELETE FROM rolepermission WHERE id=${i}$ AND roleid='any'", {"i": rp_id})
|
||||
deleted += 1
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
left = await sor.sqlExe(
|
||||
"SELECT COUNT(*) AS c FROM rolepermission WHERE roleid='any'", {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
print(f"已删除 {deleted} 条；剩余 any 授权 {getattr(left[0], 'c', '?')} 条")
|
||||
print("请执行: redis-cli DEL sc:rbac:role_perms 然后重启服务")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
ap = argparse.ArgumentParser(description="清理指向不存在路径的 any 脏授权")
|
||||
ap.add_argument('--apply', action='store_true', help="实际执行删除（默认只预演）")
|
||||
a = ap.parse_args()
|
||||
sys.exit(asyncio.run(main(a.apply)) or 0)
|
||||
Loading…
x
Reference in New Issue
Block a user