pipeline-app/scripts/clean_dirty_perms.py

#!/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)