92 lines
3.5 KiB
Python
Executable File
92 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Static audit for fragile SQLAlchemy text() patterns in ClientFlow.
|
|
|
|
This is intentionally conservative: it does not prove a bug, it flags code that
|
|
should be reviewed before deploy because PostgreSQL may be unable to infer bind
|
|
parameter types in JSONB/NULL/ANY contexts.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCAN_DIRS = [ROOT / "app", ROOT / "scripts"]
|
|
|
|
PATTERNS = [
|
|
("jsonb_build_object_with_bind", re.compile(r"jsonb_build_object\s*\([^)]*:[a-zA-Z_][a-zA-Z0-9_]", re.I | re.S), "Prefer metadata || CAST(:metadata_patch AS jsonb) or cast each bind explicitly."),
|
|
("bind_is_null_without_cast", re.compile(r"\([^)]*:[a-zA-Z_][a-zA-Z0-9_]*\s+IS\s+NULL", re.I | re.S), "Use CAST(:param AS text/uuid/...) IS NULL when parameter can be null."),
|
|
("any_bind_array", re.compile(r"\bANY\s*\(\s*:[a-zA-Z_][a-zA-Z0-9_]*\s*\)", re.I), "Prefer expanding bindparams or cast array type explicitly."),
|
|
("metadata_jsonb_concat_without_cast", re.compile(r"metadata\s*=\s*COALESCE\([^\n]+\)\s*\|\|\s*(?!CAST\(:metadata_patch\s+AS\s+JSONB\))", re.I), "Prefer COALESCE(metadata,'{}'::jsonb) || CAST(:metadata_patch AS jsonb)."),
|
|
]
|
|
|
|
ALLOW_HINTS = [
|
|
"CAST(:",
|
|
"CAST(%(",
|
|
"CAST(:metadata_patch AS JSONB)",
|
|
"CAST(:metadata_patch AS jsonb)",
|
|
]
|
|
|
|
|
|
def iter_files():
|
|
for base in SCAN_DIRS:
|
|
if not base.exists():
|
|
continue
|
|
for path in base.rglob("*.py"):
|
|
if "__pycache__" in path.parts:
|
|
continue
|
|
yield path
|
|
|
|
|
|
def line_no(text: str, pos: int) -> int:
|
|
return text.count("\n", 0, pos) + 1
|
|
|
|
|
|
def excerpt(text: str, pos: int, size: int = 240) -> str:
|
|
start = max(0, pos - 80)
|
|
end = min(len(text), pos + size)
|
|
return re.sub(r"\s+", " ", text[start:end]).strip()
|
|
|
|
|
|
def main() -> int:
|
|
findings: List[Dict[str, str]] = []
|
|
for path in iter_files():
|
|
text = path.read_text(errors="ignore")
|
|
for code, pattern, advice in PATTERNS:
|
|
for match in pattern.finditer(text):
|
|
sample = excerpt(text, match.start())
|
|
# If the local expression already has explicit casts in the same
|
|
# small sample, lower severity but still report as review.
|
|
severity = "WARN" if any(h in sample for h in ALLOW_HINTS) else "REVIEW"
|
|
findings.append({
|
|
"severity": severity,
|
|
"code": code,
|
|
"file": str(path.relative_to(ROOT)),
|
|
"line": str(line_no(text, match.start())),
|
|
"excerpt": sample,
|
|
"advice": advice,
|
|
})
|
|
print("ClientFlow SQL safety static audit")
|
|
print("=" * 80)
|
|
print(f"findings: {len(findings)}")
|
|
for item in findings[:80]:
|
|
print(f"{item['severity']} | {item['code']} | {item['file']}:{item['line']}")
|
|
print(f" {item['excerpt']}")
|
|
print(f" advice: {item['advice']}")
|
|
if len(findings) > 80:
|
|
print(f"... {len(findings) - 80} more findings not printed")
|
|
out_dir = ROOT / "audit_reports"
|
|
out_dir.mkdir(exist_ok=True)
|
|
out = out_dir / "sql_safety_findings.json"
|
|
out.write_text(json.dumps(findings, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"JSON: {out}")
|
|
# Static findings are warnings by design, not deployment blockers.
|
|
return 1 if findings else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|