diff --git a/scripts/clean_dirty_perms.py b/scripts/clean_dirty_perms.py
new file mode 100644
index 0000000..8251f26
--- /dev/null
+++ b/scripts/clean_dirty_perms.py
@@ -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)