llm-security/scanners/ast-taint-scanner.mjs

// 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.',
  });
}