feat(llm-security): add AST Python-taint scanner with python3 fallback

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V3s6WnubSSrFjAQTLQdVbG

scanners/lib/py-ast-taint.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python3
+"""py-ast-taint.py — PARSE-ONLY Python taint helper for the AST scanner.
+
+Reads a single target .py path (argv[1]), parses it with ast.parse, and walks
+the tree for variable-level taint (data flow from an input source to a
+dangerous sink) within each function/module scope. It prints one JSON object
+to stdout:
+
+    {"status": "ok", "findings": [
+        {"rule": "...", "severity": "...", "line": int,
+         "source": "...", "sink": "...", "message": "..."}]}
+
+On a parse error it prints {"status": "error", "message": "..."} and exits 2.
+
+SAFETY INVARIANT: this helper only PARSES the target (ast.parse). It never
+exec/eval/compile/imports the target, so analysing hostile code has no side
+effects. The only filesystem access is reading the target as text.
+"""
+
+import ast
+import json
+import sys
+
+# Sinks: dotted name (or bare builtin) -> (rule, severity, category)
+CODE_EXEC_SINKS = {
+    "exec": ("AST-CODE-EXEC", "critical"),
+    "eval": ("AST-CODE-EXEC", "critical"),
+    "os.system": ("AST-CMD-EXEC", "critical"),
+    "os.popen": ("AST-CMD-EXEC", "critical"),
+}
+NET_SINKS = {
+    "requests.post": ("AST-NET-EXFIL", "critical"),
+    "requests.put": ("AST-NET-EXFIL", "critical"),
+    "requests.patch": ("AST-NET-EXFIL", "critical"),
+}
+
+READ_SOURCE_CALLS = {
+    "os.getenv": "os.getenv",
+    "input": "input",
+    "requests.get": "requests.get",
+    "requests.request": "requests.request",
+}
+
+
+def dotted(node):
+    """Return the dotted name for a Name/Attribute chain, else None."""
+    if isinstance(node, ast.Name):
+        return node.id
+    if isinstance(node, ast.Attribute):
+        base = dotted(node.value)
+        return base + "." + node.attr if base else None
+    return None
+
+
+def open_mode(call):
+    """Best-effort 'mode' string of an open(...) call (default 'r')."""
+    if len(call.args) >= 2 and isinstance(call.args[1], ast.Constant):
+        return str(call.args[1].value)
+    for kw in call.keywords:
+        if kw.arg == "mode" and isinstance(kw.value, ast.Constant):
+            return str(kw.value.value)
+    return "r"
+
+
+def is_write_open(call):
+    return any(c in open_mode(call) for c in ("w", "a", "x", "+"))
+
+
+def source_label(value):
+    """If `value` is a taint source expression, return a label, else None."""
+    if isinstance(value, ast.Subscript):
+        if dotted(value.value) == "os.environ":
+            return "os.environ"
+        return source_label(value.value)
+    if isinstance(value, ast.Attribute):
+        d = dotted(value)
+        if d == "os.environ":
+            return "os.environ"
+        if d and d.startswith("sys.stdin"):
+            return "sys.stdin"
+        return None
+    if isinstance(value, ast.Call):
+        fd = dotted(value.func)
+        if fd in READ_SOURCE_CALLS:
+            return READ_SOURCE_CALLS[fd]
+        if fd == "open":
+            return None if is_write_open(value) else "open"
+        if fd and fd.startswith("sys.stdin"):
+            return "sys.stdin"
+        # Chained access on a source, e.g. open(p).read() or sys.stdin.read()
+        if isinstance(value.func, ast.Attribute):
+            return source_label(value.func.value)
+    return None
+
+
+def assigned_names(target):
+    """Yield bound Name ids for an assignment target (Name / Tuple / List)."""
+    if isinstance(target, ast.Name):
+        yield target.id
+    elif isinstance(target, (ast.Tuple, ast.List)):
+        for elt in target.elts:
+            yield from assigned_names(elt)
+
+
+def walk_scope(body):
+    """Pre-order walk of a scope's nodes, treating a nested function/class/lambda
+    as opaque (its body belongs to a separate scope and is analysed on its own)."""
+    stack = list(body)
+    while stack:
+        node = stack.pop(0)
+        yield node
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)):
+            continue  # nested scope — do not descend
+        stack[:0] = list(ast.iter_child_nodes(node))
+
+
+def tainted_arg(call, tainted):
+    """Return (name, info) for the first directly-passed tainted Name arg."""
+    candidates = list(call.args) + [kw.value for kw in call.keywords]
+    for arg in candidates:
+        if isinstance(arg, ast.Name) and arg.id in tainted:
+            return arg.id, tainted[arg.id]
+    return None
+
+
+def sink_for(call, write_handles):
+    """Return (rule, severity, sink_label) if `call` is a sink, else None."""
+    fd = dotted(call.func)
+    if fd in CODE_EXEC_SINKS:
+        rule, sev = CODE_EXEC_SINKS[fd]
+        return rule, sev, fd
+    if fd in NET_SINKS:
+        rule, sev = NET_SINKS[fd]
+        return rule, sev, fd
+    if fd and fd.startswith("subprocess."):
+        return "AST-CMD-EXEC", "critical", fd
+    if isinstance(call.func, ast.Attribute) and call.func.attr == "write":
+        recv = call.func.value
+        if isinstance(recv, ast.Name) and recv.id in write_handles:
+            return "AST-FILE-WRITE", "high", "file.write"
+    return None
+
+
+def analyze_scope(body):
+    tainted = {}          # name -> (lineno, source_label)
+    write_handles = set()  # names bound to open(..., 'w'/'a')
+    findings = []
+    for node in walk_scope(body):
+        if isinstance(node, ast.Assign):
+            src = source_label(node.value)
+            if src:
+                for name in (n for t in node.targets for n in assigned_names(t)):
+                    tainted[name] = (node.lineno, src)
+            if isinstance(node.value, ast.Call) and dotted(node.value.func) == "open" \
+                    and is_write_open(node.value):
+                for name in (n for t in node.targets for n in assigned_names(t)):
+                    write_handles.add(name)
+        if isinstance(node, ast.Call):
+            sink = sink_for(node, write_handles)
+            if sink:
+                hit = tainted_arg(node, tainted)
+                if hit:
+                    name, (src_line, src_label) = hit
+                    rule, sev, sink_label = sink
+                    findings.append({
+                        "rule": rule,
+                        "severity": sev,
+                        "line": getattr(node, "lineno", 0),
+                        "source": src_label,
+                        "sink": sink_label,
+                        "message": (
+                            "Tainted value from %s (line %d) reaches %s via `%s`."
+                            % (src_label, src_line, sink_label, name)
+                        ),
+                    })
+    return findings
+
+
+def iter_scopes(tree):
+    yield tree.body
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            yield node.body
+
+
+def main():
+    if len(sys.argv) < 2:
+        print(json.dumps({"status": "error", "message": "missing target path"}))
+        sys.exit(2)
+    path = sys.argv[1]
+    try:
+        with open(path, "r", encoding="utf-8", errors="replace") as fh:
+            src = fh.read()
+    except OSError as exc:
+        print(json.dumps({"status": "error", "message": "read error: %s" % exc}))
+        sys.exit(2)
+    try:
+        tree = ast.parse(src)
+    except (SyntaxError, ValueError) as exc:
+        print(json.dumps({"status": "error", "message": "parse error: %s" % exc}))
+        sys.exit(2)
+
+    findings = []
+    seen = set()
+    for body in iter_scopes(tree):
+        for f in analyze_scope(body):
+            key = (f["rule"], f["line"], f["source"], f["sink"])
+            if key in seen:
+                continue
+            seen.add(key)
+            findings.append(f)
+    print(json.dumps({"status": "ok", "findings": findings}))
+
+
+if __name__ == "__main__":
+    main()