# SQLi & XSS Payload Sanitizer > Identifies unescaped user inputs and unparameterized database queries, injecting sanitization middleware and parameterized bindings to neutralize XSS and SQLi. - Skill: `rmazrim/sqli-xss-payload-sanitizer` (Agent Skill) - Install (CLI): `npx skillmds@latest add rmazrim/sqli-xss-payload-sanitizer` - Raw SKILL.md: https://api.skillmd.com/api/skills/rmazrim/sqli-xss-payload-sanitizer/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Security - Author: RMAzrim (https://skillmd.com/u/rmazrim) - Updated: 2026-09-22 - Page: https://skillmd.com/skills/rmazrim/sqli-xss-payload-sanitizer --- # SQLi & XSS Payload Sanitizer ## 1. System Architecture & Prerequisites - Python 3.10+ (stdlib only: `re`, `os`, `json`, `argparse`, `shutil`, `pathlib`) - Target: local source tree with `.py`, `.js`, `.ts`, `.jsx`, `.tsx`, `.html`, `.j2`, `.jinja`, `.jinja2` files - No external dependencies ## 2. Input/Output Data Contracts **Input (CLI args):** ```json { "type": "object", "properties": { "target": { "type": "string", "description": "Root of source tree to scan/sanitize" }, "report_dir": { "type": "string", "description": "Output directory", "default": "./sanitize-reports" }, "fix": { "type": "boolean", "description": "Write sanitized files to output tree", "default": false } }, "required": ["target"] } ``` **Output artifacts:** - `{report_dir}/sanitizer_report.json` — findings with per-finding patches - `{report_dir}/sanitized/` — rewritten sanitized files (when `--fix`) - `{report_dir}/backups/` — original files (rollback source) ## 3. Production Reference Implementation ```python #!/usr/bin/env python3 """SQLi & XSS Payload Sanitizer — sink classification, sanitizer patches, AST+regex engine.""" import re import os import json import shutil import argparse from pathlib import Path # ─── Sink Classifiers ─────────────────────────────────────────────────────────── SQL_SINKS = [ re.compile(r"""(?:cursor\.execute|cursor\.executemany|\.execute\(|conn\.execute|\.raw\(|= conn\.cursor\(\)\.execute|query\()""", re.IGNORECASE), re.compile(r"""(?:SELECT|INSERT|UPDATE|DELETE|DROP|ALTER)\s+""", re.IGNORECASE), re.compile(r"""(?:\.query\(|\.exec\(|db_session\.execute|Session\.execute)""", re.IGNORECASE), ] SQL_INTERPOLATION = [ re.compile(r"""(?:execute|query|exec)\s*\(\s*f["']""", re.IGNORECASE | re.MULTILINE), re.compile(r"""(?:execute|query|exec)\s*\(\s*["'][^"']*["']\s*%""", re.IGNORECASE | re.MULTILINE), re.compile(r"""(?:execute|query|exec)\s*\(\s*["'][^"']*["']\s*\+""", re.IGNORECASE | re.MULTILINE), re.compile(r"""(?:SELECT\s+[^"]*?["']?\s*(\{|%s|\%(?:\w+\()|\.format\())""", re.IGNORECASE | re.MULTILINE), re.compile(r"""(?:f["'][^"']*(?:\{[^}]+\})[^"']*["'](?:\s*\)))""", re.MULTILINE), ] XSS_SINKS = [ ("innerHTML", re.compile(r"""\.innerHTML\s*="""), "innerHTML assignment"), ("insertAdjacentHTML", re.compile(r"""insertAdjacentHTML\s*\("""), "insertAdjacentHTML call"), ("dangerouslySetInnerHTML", re.compile(r"""dangerouslySetInnerHTML\s*=""", re.MULTILINE), "React dangerouslySetInnerHTML"), ("safe_filter", re.compile(r"""\|\s*safe(?:\s|}}|\b)""", re.MULTILINE), "Jinja2 |safe filter"), ("v-html", re.compile(r"""v-html\s*="""), "Vue v-html directive"), ("document.write", re.compile(r"""document\.write\s*\("""), "document.write call"), ("innerHTML_var", re.compile(r"""\.innerHTML\s*=\s*[a-zA-Z_$]"""), "innerHTML assigned from variable"), ] USER_INPUT_SOURCES = [ re.compile(r"""(?:request\.(?:form|args|params|json|values)|req\.body|req\.query|req\.params|input\(|getParameter|body\(\s*\)|@request\.body|cgi\.FieldStorage|environ\[)""", re.IGNORECASE), re.compile(r"""(?:getElementById|querySelector|login|signup|comments?|search|username|email|password|name\b)""", re.IGNORECASE), ] HTML_ESCAPE_TABLE = { "&": "&", "<": "<", ">": ">", '"': """, "'": "'", "/": "/", "`": "`", "=": "=", } def html_escape(text: str) -> str: """OWASP-recommended HTML escape. Replaces &<>\"'/`= with entities.""" return "".join(HTML_ESCAPE_TABLE.get(c, c) for c in text) # ─── Analysis Engine ──────────────────────────────────────────────────────────── SKIP_DIRS = {"node_modules", ".git", "__pycache__", "venv", ".venv", "dist", "build", ".next"} SOURCE_EXTS = {".py", ".js", ".ts", ".jsx", ".tsx", ".html", ".j2", ".jinja", ".jinja2"} def classify_line(line: str) -> dict: """Classify a single source line. Returns finding dict or None.""" finding = {"line_text": line.strip()[:200]} has_sql_sink = any(p.search(line) for p in SQL_SINKS) has_interp = any(p.search(line) for p in SQL_INTERPOLATION) user_input_nearby = any(p.search(line) for p in USER_INPUT_SOURCES) if has_sql_sink and (has_interp or user_input_nearby): finding.update({ "type": "SQLI", "cwe": "CWE-89", "severity": "CRITICAL", "detail": "Unparameterized SQL with interpolation; user input may be concatenated into query", "patch_kind": "parameterize", }) return finding for sink_id, pattern, desc in XSS_SINKS: if pattern.search(line): finding.update({ "type": "XSS", "cwe": "CWE-79", "severity": "HIGH", "detail": f"Unescaped output sink: {desc}", "patch_kind": "escape", "sink_id": sink_id, }) return finding return None def scan_source(root: Path) -> tuple: """Scan all source files. Returns (findings, files_scanned).""" findings = [] files_scanned = 0 for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS] for fname in filenames: fpath = Path(dirpath) / fname if fpath.suffix not in SOURCE_EXTS: continue files_scanned += 1 try: lines = fpath.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: continue for i, line in enumerate(lines): finding = classify_line(line) if finding: finding["file"] = str(fpath) finding["line"] = i + 1 findings.append(finding) return findings, files_scanned # ─── Patch Engine ─────────────────────────────────────────────────────────────── def parameterize_sql(line: str, connector: str = " ") -> str: """Rewrite an f-string/concatenated SQL line to use placeholders + param array.""" m = re.search(r"""(execute|query|exec)\s*\(\s*f(["'])(SELECT\s+.*?)\2\s*\)""", line, re.IGNORECASE | re.DOTALL) if m: call = m.group(1) quote = m.group(2) sql = m.group(3) params = re.findall(r"\{(\w+)\}", sql) clean = re.sub(r"\{(\w+)\}", "%s", sql) return f'{call}{connector}({quote}{clean}{quote}, ({", ".join(params)},))' # positional style: execute("... %s ..." % (x,)) -> execute("... %s ...", (x,)) m = re.search(r"""(execute|query|exec)\s*\(\s*(["'])(.*?)\2\s*%\s*\(([^)]*)\)\s*\)""", line, re.IGNORECASE | re.DOTALL) if m: call, quote, sql, param_tuple = m.group(1), m.group(2), m.group(3), m.group(4) return f'{call}{connector}({quote}{sql}{quote}, ({param_tuple},))' return line def escape_html(line: str, sink_id: str) -> str: """Inject htmlEscape wrapper around the sink payload.""" if sink_id == "innerHTML": return re.sub(r"""\.innerHTML(\s*=\s*)([^;]+)""", lambda m: f".textContent{m.group(1)}htmlEscape({m.group(2).strip()})", line) if sink_id == "insertAdjacentHTML": return re.sub(r"""insertAdjacentHTML\s*\(\s*([^,]+),\s*([^)]+)\)""", lambda m: f"insertAdjacentHTML({m.group(1).strip()}, htmlEscape({m.group(2).strip()}))", line) if sink_id == "dangerouslySetInnerHTML": return re.sub(r"""dangerouslySetInnerHTML(\s*=\s*)\{\s*__html:\s*([^}]+)\}""", lambda m: f"dangerouslySetInnerHTML{m.group(1)}{{ __html: htmlEscape({m.group(2).strip()}) }}", line) if sink_id == "safe_filter": return line.replace("| safe", "| html_escape") if "| safe" in line else line.replace("|safe", "|html_escape") if sink_id == "v-html": return line.replace("v-html=", ":v-text=") if sink_id == "document.write": return re.sub(r"""document\.write\s*\(\s*([^)]+)\)""", lambda m: f"document.getElementById('output').textContent = htmlEscape({m.group(1).strip()})", line) return line ESCAPE_UTIL_JS = """// htmlEscape — OWASP/DOMPurify-safe HTML encoder function htmlEscape(text) { return String(text) .replace(/&/g, '&') .replace(//g, '>') .replace(/"/g, '"') .replace(/'/g, ''') .replace(/\\//g, '/') .replace(/`/g, '`') .replace(/=/g, '='); } if (typeof module !== 'undefined' && module.exports) module.exports = { htmlEscape }; """ ESCAPE_UTIL_PY = '''# html_escape — OWASP-recommended HTML encoder _ESCAPE_TABLE = { "&": "&", "<": "<", ">": ">", '"': """, "'": "'", "/": "/", "`": "`", "=": "=", } def html_escape(text) -> str: if text is None: return "" return "".join(_ESCAPE_TABLE.get(ch, ch) for ch in str(text)) ''' def sanitize_file(fpath: Path, root: Path, report_dir: Path) -> list: """Rewrite a file applying param/escape patches. Returns list of applied patches.""" content = fpath.read_text(encoding="utf-8", errors="replace") lines = content.splitlines(keepends=True) patches = [] for i, line in enumerate(lines): finding = classify_line(line) if not finding: continue patch = None if finding["type"] == "SQLI" and finding["patch_kind"] == "parameterize": patch = parameterize_sql(line, connector="") elif finding["type"] == "XSS": patch = escape_html(line, finding.get("sink_id", "")) if patch and patch != line: if not patch.endswith("\n") and line.endswith("\n"): patch = patch + "\n" lines[i] = patch patches.append({ "file": str(fpath), "line": i + 1, "type": finding["type"], "before": line.strip()[:200], "after": patch.strip()[:200], }) if patches: rel = fpath.relative_to(root) out = report_dir / "sanitized" / rel out.parent.mkdir(parents=True, exist_ok=True) out.write_text("".join(lines), encoding="utf-8") backup = report_dir / "backups" / rel backup.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(fpath, backup) if fpath.suffix in {".js", ".ts", ".jsx", ".tsx"}: util = report_dir / "sanitized" / "_html_escape.js" if not util.exists(): util.write_text(ESCAPE_UTIL_JS, encoding="utf-8") elif fpath.suffix == ".py": util = report_dir / "sanitized" / "_html_escape.py" if not util.exists(): util.write_text(ESCAPE_UTIL_PY, encoding="utf-8") return patches # ─── Orchestration ────────────────────────────────────────────────────────────── def console_table(findings: list, files_scanned: int) -> str: by_type = {} for f in findings: by_type[f["type"]] = by_type.get(f["type"], 0) + 1 table = ["SQLi / XSS Sanitizer — scan summary", ""] table.append(f"Files scanned: {files_scanned}") table.append(f"Total findings: {len(findings)}") table.append(f" SQLI (CWE-89): {by_type.get('SQLI', 0)}") table.append(f" XSS (CWE-79): {by_type.get('XSS', 0)}") return "\n".join(table) def main(): parser = argparse.ArgumentParser( description="SQLi & XSS Payload Sanitizer — classify sinks, emit patches, rewrite secure code.", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--target", required=True, help="Root of source tree to scan/sanitize") parser.add_argument("--report-dir", default="./sanitize-reports", help="Output directory") parser.add_argument("--fix", action="store_true", help="Write sanitized files + backups") args = parser.parse_args() root = Path(args.target).resolve() if not root.is_dir(): print(f"[ERROR] Target directory does not exist: {root}") raise SystemExit(1) report_dir = Path(args.report_dir).resolve() report_dir.mkdir(parents=True, exist_ok=True) print(f"[INFO] Scanning: {root}") findings, files_scanned = scan_source(root) print(console_table(findings, files_scanned)) for f in findings: print(f" [{f['severity']}] {f['type']} {f['file']}:{f['line']} — {f['detail']}") applied = [] if args.fix: print("[INFO] Applying sanitizer patches...") seen = set() for f in findings: fpath = Path(f["file"]) if fpath in seen: continue seen.add(fpath) applied.extend(sanitize_file(fpath, root, report_dir)) print(f"[INFO] Patched {len(applied)} line(s) across {len({p['file'] for p in applied})} file(s)") report = { "target": str(root), "files_scanned": files_scanned, "findings": findings, "patches_applied": applied, "output": { "sanitized_tree": str(report_dir / "sanitized"), "backups": str(report_dir / "backups"), }, "rollback": "Copy original files from backups/ back to source locations.", } report_path = report_dir / "sanitizer_report.json" report_path.write_text(json.dumps(report, indent=2, default=str), encoding="utf-8") print(f"\n[INFO] Report: {report_path}") print(f"[INFO] If --fix used: review {report_dir / 'sanitized'}; rollback from {report_dir / 'backups'}") if __name__ == "__main__": main() ``` ## 4. Execution Protocol & Step-by-Step Workflow 1. Run a read-only classification pass: ```bash python sqli-xss-payload-sanitizer.md --target ./app --report-dir ./sanitize-reports ``` 2. Review the console table (SQLI vs XSS counts) and per-finding detail lines. 3. Inspect `sanitize-reports/sanitizer_report.json` for structured findings with patch kind (`parameterize` / `escape`). 4. Apply patches with `--fix`: ```bash python sqli-xss-payload-sanitizer.md --target ./app --report-dir ./sanitize-reports --fix ``` 5. Copy generated `_html_escape.js` / `_html_escape.py` utilities into your app and import them. 6. Review all files under `sanitize-reports/sanitized/`; original files are untouched and duplicated at `sanitize-reports/backups/`. 7. Promote sanitized files into the source tree; if anything breaks, roll back from `backups/`. 8. Re-run the scanner; new findings should be zero. ## 5. Edge Cases & Error Handling - Non-UTF-8 files are read with `errors="replace"`, so bytes are never fatal; findings only flag decodable content. - SQL rewriting keeps original line endings; JSON payloads / multiline strings are only rewritten when the sink pattern matches the exact `execute("...", (..))` or f-string form. - `--fix` never modifies source files in place — it writes to `sanitized/` plus a full `backups/` tree, guaranteeing rollback by copying backups back. - Escaping uses the full OWASP entity table (`&<>"'/\`=`), not just `&<>`; the `htmlEscape`/`html_escape` utilities are generated alongside patched code so imports resolve. - Pattern engine is regex-based (not full AST); false positives on dynamic SQL builders (e.g. ORM `.query()` chains) are possible — severity flags recommend human review. - Files with binary content that match source extensions decode via `errors="replace"` and are ignored if no sink patterns match.