pbl_blueprint/scripts/audit_dspy.py

58 lines
2.4 KiB
Python

#!/usr/bin/env python3
"""dspy 禁项审计 + sqlor API 白名单 + api 文件/RBAC 注册对齐检查。"""
import os
import re
import sys
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
API_DIR = os.path.join(ROOT, 'pbl_blueprint', 'wwwroot', 'api')
BANNED = [
('import ', r'(?m)^\s*(import|from)\s+\w'),
('f-string', r"(?<![\w])f['\"]"),
('print(', r'\bprint\s*\('),
('uuid', r'\buuid\b'),
('os.environ', r'os\.environ'),
('open(', r'\bopen\s*\('),
]
SOR_ALLOWED = {'C', 'U', 'D', 'R', 'I', 'sqlExe'}
def audit():
problems, files = [], sorted(f for f in os.listdir(API_DIR) if f.endswith('.dspy'))
for name in files:
raw = open(os.path.join(API_DIR, name), encoding='utf-8').read()
# 注释行不参与禁项判定(禁项针对可执行代码)
text = '\n'.join(l for l in raw.splitlines() if not l.strip().startswith('#'))
for label, pat in BANNED:
for m in re.finditer(pat, text):
problems.append('%s: 禁项 %s @%s' % (name, label, m.group(0).strip()))
for m in re.finditer(r'\bsor\.(\w+)', text):
if m.group(1) not in SOR_ALLOWED:
problems.append('%s: 非法 sqlor API sor.%s' % (name, m.group(1)))
for m in re.finditer(r'\bsor\.(\w+)\s*\(', text):
pass
if 'return ' not in text:
problems.append('%s: 缺少显式 return' % name)
sys.path.insert(0, os.path.join(ROOT, 'scripts'))
import load_path
registered = {p['path'].split('/%s/' % load_path.MODULE, 1)[1] for p in load_path.PATHS}
for name in files:
if 'api/' + name not in registered:
problems.append('%s: 未在 scripts/load_path.py 注册' % name)
for p in sorted(registered):
if p.startswith('api/') and not os.path.exists(os.path.join(API_DIR, os.path.basename(p))):
problems.append('注册了不存在的文件: %s' % p)
for ui in sorted(os.listdir(os.path.join(ROOT, 'pbl_blueprint', 'wwwroot'))):
if ui.endswith('.ui') and ui not in registered:
problems.append('%s: 未在 scripts/load_path.py 注册' % ui)
return files, problems
if __name__ == '__main__':
files, problems = audit()
print('audited dspy files: %d' % len(files))
for p in problems:
print('FAIL ' + p)
print('DSPY_AUDIT_%s' % ('PASS' if not problems else 'FAIL(%d)' % len(problems)))
sys.exit(1 if problems else 0)