60 lines
3.6 KiB
JavaScript
60 lines
3.6 KiB
JavaScript
import test from 'node:test'
|
|
import assert from 'node:assert/strict'
|
|
import fs from 'node:fs'
|
|
import os from 'node:os'
|
|
import path from 'node:path'
|
|
import { execFileSync } from 'node:child_process'
|
|
import * as plugin from '../index.js'
|
|
import { SCANNERS } from '../lib/scanners.js'
|
|
|
|
function harness({ llmText } = {}) {
|
|
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'omp-sec-'))
|
|
const tools = new Map()
|
|
const llm = {
|
|
async *stream() { yield { type: 'text-delta', text: llmText ?? '{"triage":[],"new":[],"summary":"stub"}' }; yield { type: 'finish', reason: { kind: 'stop' } } },
|
|
}
|
|
plugin.apply({ tools: { register(d) { tools.set(d.name, d) } }, effect() {}, llm }, {})
|
|
const exec = { agent: { session: { header: { cwd: dir }, requestHeader: () => ({ config: { provider: 'p', model: 'm' } }) } } }
|
|
const call = async (args) => (await tools.get('security_scan').execute(args, exec)).text
|
|
return { dir, call, w: (f, s) => { fs.mkdirSync(path.dirname(path.join(dir, f)), { recursive: true }); fs.writeFileSync(path.join(dir, f), s) } }
|
|
}
|
|
|
|
const vulnPy = `import subprocess, pickle, hashlib\n\ndef run(cmd):\n return subprocess.call(cmd, shell=True)\n\ndef load(b):\n return pickle.loads(b)\n\ndef h(x):\n return hashlib.md5(x).hexdigest()\n`
|
|
const secret = 'AWS_SECRET = "' + 'AKIA' + 'Z7QWERTYUIOPASDF' + '"\naws_secret_access_key = "' + 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY' + '"\n'
|
|
|
|
test('scanners action lists tools', async () => {
|
|
const h = harness()
|
|
assert.match(await h.call({ action: 'scanners' }), /semgrep/)
|
|
})
|
|
|
|
test('preflight → start → status finds python issues and AI triage applies', { skip: !SCANNERS.bandit.available(), timeout: 240000 }, async () => {
|
|
const llmText = JSON.stringify({ triage: [], new: [{ file: 'app.py', line: 4, severity: 'high', title: 'Command injection via shell=True', reason: 'cmd reaches a shell', fix: 'pass a list and shell=False', cwe: 'CWE-78' }], summary: 'Fix the shell injection first.' })
|
|
const h = harness({ llmText })
|
|
h.w('app.py', vulnPy)
|
|
h.w('creds.py', secret)
|
|
execFileSync('git', ['init', '-q'], { cwd: h.dir })
|
|
const pf = await h.call({ action: 'preflight', target_kind: 'scoped_path', include_paths: ['app.py', 'creds.py'], scanners: ['bandit', 'gitleaks'] })
|
|
const planId = /Plan (plan-\w+)/.exec(pf)[1]
|
|
assert.match(pf, /✓ bandit/)
|
|
const st = await h.call({ action: 'start', plan_id: planId, wait_seconds: 200 })
|
|
assert.match(st, /: completed;|: partial;/, st)
|
|
assert.match(st, /B602|subprocess.*shell/i, st)
|
|
assert.match(st, /pickle/i)
|
|
assert.match(st, /AI review \(p\/m\): Fix the shell injection first/)
|
|
assert.match(st, /ai-review/)
|
|
const scanId = /Security scan (scan-[\w-]+)/.exec(st)[1]
|
|
const fidm = /id (\w{10})/.exec(st)[1]
|
|
assert.match(await h.call({ action: 'validate', scan_id: scanId, finding_id: fidm, validation_status: 'validated', validation_summary: 'confirmed by test' }), /validated/)
|
|
assert.match(await h.call({ action: 'status', scan_id: scanId }), /Validation: validated/)
|
|
fs.rmSync(path.join(os.homedir(), '.dsh/security-scans', `${scanId}.json`), { force: true })
|
|
})
|
|
|
|
test('osv-scanner flags a vulnerable lockfile', { skip: !SCANNERS['osv-scanner'].available(), timeout: 240000 }, async () => {
|
|
const h = harness()
|
|
h.w('requirements.txt', 'django==2.2.0\n')
|
|
const pf = await h.call({ action: 'preflight', scanners: ['osv-scanner'], ai_review: false })
|
|
const st = await h.call({ action: 'start', plan_id: /Plan (plan-\w+)/.exec(pf)[1], wait_seconds: 200 })
|
|
if (/osv-scanner ERROR/.test(st)) return // offline: tolerated
|
|
assert.match(st, /django@2\.2\.0/i, st)
|
|
})
|