88 lines
4.7 KiB
Python
88 lines
4.7 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
from importers.nvd import simple_or_matches
|
|
|
|
|
|
def run(*args, cwd=None):
|
|
return subprocess.run(args, cwd=cwd, check=True, text=True, capture_output=True)
|
|
|
|
|
|
class BuilderTests(unittest.TestCase):
|
|
def test_complex_nvd_rejected(self):
|
|
config = {'operator':'OR', 'nodes':[{'operator':'AND', 'cpeMatch':[{'criteria':'cpe:2.3:a:x:y:*:*:*:*:*:*:*:*'}]}]}
|
|
self.assertEqual(list(simple_or_matches(config)), [])
|
|
|
|
def test_end_to_end(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root = Path(folder)
|
|
source_root = root / 'sources'
|
|
source_root.mkdir()
|
|
samples = {
|
|
'vuln-cve': ('data/2026/1xxx/CVE-2026-1234.json', {
|
|
'cveMetadata':{'cveId':'CVE-2026-1234','state':'PUBLISHED','dateUpdated':'2026-01-01T00:00:00Z'}}),
|
|
'vuln-nvd': ('data/2026/CVE-2026-1234.json', {
|
|
'id':'CVE-2026-1234','published':'2026-01-01T00:00:00Z',
|
|
'lastModified':'2026-11-12T00:00:00Z','vulnStatus':'Analyzed',
|
|
'configurations':[{'operator':'OR','nodes':[{'operator':'OR','cpeMatch':[
|
|
{'criteria':'cpe:2.3:a:example:product:*:*:*:*:*:*:*:*','vulnerable':True}]}]}]}),
|
|
'vuln-osv': ('data/PyPI/GHSA-aaaa-bbbb-cccc.json', {
|
|
'id':'GHSA-aaaa-bbbb-cccc','modified':'2026-11-12T00:00:00Z',
|
|
'aliases':['CVE-2026-1234'],'affected':[{'package':{'ecosystem':'PyPI','name':'example'},
|
|
'ranges':[{'type':'ECOSYSTEM','events':[{'introduced':'0'},{'fixed':'2.0'}]}]}]})
|
|
}
|
|
for name, (rel, obj) in samples.items():
|
|
repo = source_root / name
|
|
target = repo / rel
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(json.dumps(obj), encoding='utf8')
|
|
run('git', 'init', '-q', str(repo))
|
|
run('git', 'add', '.', cwd=repo)
|
|
run('git', '-c', 'user.email=test@example.invalid', '-c', 'user.name=test', 'commit', '-qm', 'initial', cwd=repo)
|
|
# Include another repository without an importer; it is tracked, not indexed.
|
|
repo = source_root / 'vuln-redhat'; repo.mkdir()
|
|
run('git', 'init', '-q', str(repo))
|
|
(repo/'README.md').write_text('placeholder')
|
|
run('git', 'add', '.', cwd=repo)
|
|
run('git', '-c', 'user.email=test@example.invalid', '-c', 'user.name=test', 'commit', '-qm', 'initial', cwd=repo)
|
|
release = root / 'releases'
|
|
cmd = [sys.executable, str(ROOT/'scripts/build.py'), '--sources', str(source_root), '--output', str(release)]
|
|
first = run(*cmd)
|
|
self.assertIn('Build complete', first.stdout)
|
|
db = sqlite3.connect(release/'master.db')
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM vulnerabilities').fetchone()[0], 2)
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM affected_packages').fetchone()[0], 1)
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM cpe_matches').fetchone()[0], 1)
|
|
self.assertEqual(db.execute("SELECT indexed,documents FROM sources WHERE name='vuln-redhat'").fetchone(), (0,0))
|
|
self.assertEqual(db.execute('PRAGMA integrity_check').fetchone()[0], 'ok')
|
|
db.close()
|
|
manifest1 = json.loads((release/'manifest.json').read_text())
|
|
second = run(*cmd)
|
|
self.assertIn('NO_CHANGES', second.stdout)
|
|
self.assertEqual(json.loads((release/'manifest.json').read_text())['version'], manifest1['version'])
|
|
# Change NVD source and verify new delta contains an upsert.
|
|
file = source_root/'vuln-nvd/data/2026/CVE-2026-1234.json'
|
|
obj = json.loads(file.read_text()); obj['vulnStatus'] = 'Modified'
|
|
file.write_text(json.dumps(obj))
|
|
run('git','add','.',cwd=source_root/'vuln-nvd')
|
|
run('git','-c','user.email=test@example.invalid','-c','user.name=test','commit','-qm','change',cwd=source_root/'vuln-nvd')
|
|
third = run(*cmd)
|
|
self.assertIn('Build complete', third.stdout)
|
|
manifest2 = json.loads((release/'manifest.json').read_text())
|
|
self.assertNotEqual(manifest1['version'], manifest2['version'])
|
|
delta = release/f"delta-{manifest1['version']}-to-{manifest2['version']}.jsonl.zst"
|
|
self.assertTrue(delta.is_file())
|
|
raw = run('zstd','-dc',str(delta)).stdout
|
|
self.assertIn('"op":"upsert"',raw)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|