114 lines
4.6 KiB
Python
114 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Build compact CVE SQLite plus optionally incremental row-level delta (CVE slice)."""
|
|
import argparse
|
|
import base64
|
|
import datetime as dt
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import zlib
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
SCHEMA = '''CREATE TABLE cve (
|
|
cve_id TEXT PRIMARY KEY,
|
|
state TEXT NOT NULL,
|
|
date_updated TEXT,
|
|
affected BLOB NOT NULL,
|
|
row_hash TEXT NOT NULL
|
|
) WITHOUT ROWID;
|
|
CREATE INDEX idx_cve_state ON cve(state);'''
|
|
|
|
def encode_record(doc):
|
|
meta = doc['cveMetadata']
|
|
cna = doc.get('containers', {}).get('cna', {})
|
|
# Preserve entire CNA affected array as structured JSON. Do NOT infer ranges.
|
|
affected = json.dumps(cna.get('affected', []), sort_keys=True, separators=(',', ':'), ensure_ascii=False).encode('utf-8')
|
|
blob = zlib.compress(affected, level=9)
|
|
row = (meta['cveId'], meta.get('state', 'UNKNOWN'), meta.get('dateUpdated'), blob)
|
|
digest = hashlib.sha256(json.dumps([row[0], row[1], row[2]], separators=(',', ':')).encode() + blob).hexdigest()
|
|
return (*row, digest)
|
|
|
|
def fingerprint(conn):
|
|
h = hashlib.sha256()
|
|
for cve_id, row_hash in conn.execute('SELECT cve_id,row_hash FROM cve ORDER BY cve_id'):
|
|
h.update(cve_id.encode() + b'\x00' + row_hash.encode() + b'\n')
|
|
return h.hexdigest()
|
|
|
|
def export_zstd(path, output):
|
|
with output.open('wb') as dst:
|
|
subprocess.run(['zstd', '-q', '-12', '-c', str(path)], stdout=dst, check=True)
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('--source', type=Path, required=True, help='vuln-cve/data')
|
|
ap.add_argument('--output', type=Path, default=Path('dist'))
|
|
ap.add_argument('--previous', type=Path, help='previous SQLite master.db')
|
|
ap.add_argument('--source-revision', default='unknown')
|
|
args = ap.parse_args()
|
|
args.output.mkdir(parents=True, exist_ok=True)
|
|
db = args.output / 'master.db'
|
|
db.unlink(missing_ok=True)
|
|
current = sqlite3.connect(db)
|
|
current.executescript(SCHEMA)
|
|
count = 0
|
|
for path in sorted(args.source.rglob('CVE-*.json')):
|
|
with path.open(encoding='utf-8') as fh:
|
|
doc = json.load(fh)
|
|
if doc.get('cveMetadata', {}).get('cveId') != path.stem:
|
|
raise ValueError(f'CVE metadata mismatch at {path}')
|
|
current.execute('INSERT INTO cve VALUES (?,?,?,?,?)', encode_record(doc))
|
|
count += 1
|
|
if count % 5000 == 0:
|
|
current.commit()
|
|
if count == 0:
|
|
raise RuntimeError('No CVE records, refusing empty release')
|
|
current.commit()
|
|
current.execute('VACUUM')
|
|
current.commit()
|
|
current_hash = fingerprint(current)
|
|
previous_hash = None
|
|
changed = removed = 0
|
|
if args.previous:
|
|
prev = sqlite3.connect(f'file:{args.previous.resolve()}?mode=ro', uri=True)
|
|
previous_hash = fingerprint(prev)
|
|
old = {row[0]: row[1] for row in prev.execute('SELECT cve_id,row_hash FROM cve')}
|
|
delta_file = args.output / 'delta.jsonl.zst'
|
|
plain_delta = args.output / 'delta.jsonl'
|
|
with plain_delta.open('w', encoding='utf-8') as out:
|
|
header = {'type': 'header', 'format': 1, 'base': previous_hash, 'target': current_hash}
|
|
out.write(json.dumps(header) + '\n')
|
|
for id, state, date, blob, row_hash in current.execute('SELECT * FROM cve ORDER BY cve_id'):
|
|
old_hash = old.pop(id, None)
|
|
if old_hash != row_hash:
|
|
change = {'type': 'upsert', 'id': id, 'state': state, 'updated': date, 'affected_b64': base64.b64encode(blob).decode(), 'hash': row_hash}
|
|
out.write(json.dumps(change, separators=(',', ':')) + '\n')
|
|
changed += 1
|
|
for id in sorted(old):
|
|
out.write(json.dumps({'type': 'delete', 'id': id}) + '\n')
|
|
removed += 1
|
|
export_zstd(plain_delta, delta_file)
|
|
plain_delta.unlink()
|
|
prev.close()
|
|
export_zstd(db, args.output / 'master.db.zst')
|
|
manifest = {
|
|
'format': 1,
|
|
'schema': 'cve-index-v1',
|
|
'built_at': dt.datetime.now(dt.timezone.utc).isoformat(),
|
|
'sources': {'cve': args.source_revision},
|
|
'records': count,
|
|
'content_sha256': current_hash,
|
|
'previous_content_sha256': previous_hash,
|
|
'upserts': changed,
|
|
'deletions': removed,
|
|
'artifacts': {},
|
|
}
|
|
for f in sorted(args.output.glob('*.zst')):
|
|
manifest['artifacts'][f.name] = {'sha256': hashlib.sha256(f.read_bytes()).hexdigest(), 'bytes': f.stat().st_size}
|
|
(args.output / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
|
|
print(json.dumps(manifest, indent=2))
|
|
current.close()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|