Merge Gitea changes into main

This commit is contained in:
root committed 2026-11-12 21:15:58 +01:00
commit 24325d8953
4 files changed
+577 -57

No files matched your search

+263 -8
View File
@@ -1,47 +1,302 @@
name: Build SQLite vulnerability index
on:
workflow_dispatch:
schedule:
- cron: "35 3 * * *"
concurrency:
group: vuln-builder
cancel-in-progress: false
jobs:
build:
runs-on: vuln-builder
timeout-minutes: 720
env:
CACHE_ROOT: /vuln-cache/vuln-builder
GITEA_BASE_URL: https://git.apex-project.de
GITEA_JOB_TOKEN: ${{ secrets.GITEA_TOKEN }}
SOURCES_DIR: /vuln-cache/vuln-builder/sources
SOURCES_CONFIG: /vuln-cache/vuln-builder/sources.json
steps:
# ============================================
# 1. Install dependencies
# ============================================
- name: Install tools
shell: sh
run: |
set -eu
if command -v apk >/dev/null 2>&1; then
apk add --no-cache bash git python3 sqlite zstd util-linux ca-certificates coreutils
apk add --no-cache \
bash git python3 sqlite zstd \
util-linux ca-certificates coreutils
elif command -v apt-get >/dev/null 2>&1; then
apt-get update
apt-get install -y --no-install-recommends bash git python3 sqlite3 zstd util-linux ca-certificates coreutils
apt-get install -y --no-install-recommends \
bash git python3 sqlite3 zstd \
util-linux ca-certificates coreutils
else
echo 'Unsupported job image' >&2
echo "Unsupported job image" >&2
exit 1
fi
python3 --version
git --version
/usr/bin/flock --version
- name: Bootstrap and build
# ============================================
# 2. Bootstrap builder
# ============================================
- name: Bootstrap builder
shell: bash
run: |
set -Eeuo pipefail
: "${GITEA_JOB_TOKEN:?missing token}"
umask 077
: "${GITEA_JOB_TOKEN:?Missing Gitea token}"
mkdir -p "$CACHE_ROOT"
export GIT_CONFIG_COUNT=1
export GIT_CONFIG_KEY_0=http.extraHeader
export GIT_CONFIG_VALUE_0="Authorization: Basic $(printf 'oauth2:%s' "$GITEA_JOB_TOKEN" | base64 | tr -d '\n')"
if [ ! -d "$CACHE_ROOT/builder/.git" ]; then
git clone --depth 1 --branch main "$GITEA_BASE_URL/VULN/vuln-builder.git" "$CACHE_ROOT/builder"
git clone --depth 1 --branch main \
"$GITEA_BASE_URL/VULN/vuln-builder.git" \
"$CACHE_ROOT/builder"
else
git -C "$CACHE_ROOT/builder" fetch --depth 1 origin main
git -C "$CACHE_ROOT/builder" reset --hard FETCH_HEAD
git -C "$CACHE_ROOT/builder" \
fetch --depth 1 origin main
git -C "$CACHE_ROOT/builder" \
reset --hard FETCH_HEAD
fi
echo "Builder ready"
# ============================================
# 3. Download sources.json
# ============================================
- name: Load sources configuration
shell: bash
run: |
set -Eeuo pipefail
umask 077
python3 - <<'PY'
import json
import os
import urllib.request
from pathlib import Path
base = os.environ["GITEA_BASE_URL"].rstrip("/")
token = os.environ["GITEA_JOB_TOKEN"]
url = (
f"{base}/VULN/vuln-status/"
"raw/branch/main/sources.json"
)
request = urllib.request.Request(
url,
headers={"Authorization": f"token {token}"}
)
with urllib.request.urlopen(
request, timeout=120
) as response:
data = response.read()
config = json.loads(data)
if not isinstance(config.get("sources"), list):
raise ValueError("Invalid sources.json")
path = Path(os.environ["SOURCES_CONFIG"])
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
print(
f"Loaded {len(config['sources'])} entries",
flush=True
)
PY
# ============================================
# 4. Synchronize all repositories
# ============================================
- name: Synchronize vulnerability repositories
shell: bash
run: |
set -Eeuo pipefail
umask 077
export GIT_CONFIG_COUNT=1
export GIT_CONFIG_KEY_0=http.extraHeader
export GIT_CONFIG_VALUE_0="Authorization: Basic $(printf 'oauth2:%s' "$GITEA_JOB_TOKEN" | base64 | tr -d '\n')"
python3 - <<'PY'
import json
import os
import re
import subprocess
from pathlib import Path
root = Path(os.environ["CACHE_ROOT"])
sources_dir = Path(os.environ["SOURCES_DIR"])
base = os.environ["GITEA_BASE_URL"].rstrip("/")
config = json.loads(
Path(os.environ["SOURCES_CONFIG"])
.read_text(encoding="utf-8")
)
owner = config.get("owner", "VULN")
branch = config.get("branch", "main")
valid_repo = re.compile(r"^vuln-[a-z0-9-]+$")
valid_owner = re.compile(r"^[A-Za-z0-9_-]+$")
valid_branch = re.compile(r"^[A-Za-z0-9_./-]+$")
if not valid_owner.fullmatch(owner):
raise ValueError("Invalid owner")
if (
not valid_branch.fullmatch(branch)
or branch.startswith("-")
or ".." in branch
or branch.endswith("/")
):
raise ValueError("Invalid branch")
sources_dir.mkdir(parents=True, exist_ok=True)
manifest = []
seen = set()
def git(*args, cwd=None):
subprocess.run(
["git", *args],
cwd=cwd,
check=True
)
for entry in config["sources"]:
if entry.get("enabled", True) is False:
continue
name = entry["repo"]
if not valid_repo.fullmatch(name):
raise ValueError(f"Invalid repo: {name}")
if name in seen:
raise ValueError(f"Duplicate repo: {name}")
seen.add(name)
repo_dir = sources_dir / name
repo_url = f"{base}/{owner}/{name}.git"
print(f"\n=== SYNC {name} ===", flush=True)
if not (repo_dir / ".git").is_dir():
git(
"clone",
"--depth", "1",
"--single-branch",
"--branch", branch,
repo_url,
str(repo_dir)
)
else:
git(
"remote", "set-url",
"origin", repo_url,
cwd=repo_dir
)
git(
"fetch",
"--depth", "1",
"origin", branch,
cwd=repo_dir
)
git(
"reset", "--hard",
"FETCH_HEAD",
cwd=repo_dir
)
commit = subprocess.check_output(
["git", "rev-parse", "HEAD"],
cwd=repo_dir,
text=True
).strip()
manifest.append({
"repo": name,
"commit": commit,
"path": str(repo_dir)
})
print(
f"OK {name}: {commit[:12]}",
flush=True
)
if not manifest:
raise RuntimeError(
"No enabled repositories found"
)
manifest_path = root / "source-manifest.json"
temp_path = root / "source-manifest.json.tmp"
temp_path.write_text(
json.dumps(manifest, indent=2),
encoding="utf-8"
)
temp_path.replace(manifest_path)
print(
f"\nSynchronized: {len(manifest)} repositories",
flush=True
)
PY
# ============================================
# 5. Build SQLite master and delta
# ============================================
- name: Build SQLite database
shell: bash
run: |
set -Eeuo pipefail
echo "Starting SQLite Builder"
bash "$CACHE_ROOT/builder/scripts/run.sh"
# ============================================
# 6. Show results
# ============================================
- name: Show build results
if: always()
shell: sh
run: |
echo "=== Cache usage ==="
du -sh "$CACHE_ROOT" 2>/dev/null || true
echo "=== Sources ==="
ls -lah "$CACHE_ROOT/sources" \
2>/dev/null || true
echo "=== Releases ==="
ls -lah "$CACHE_ROOT/releases" \
2>/dev/null || true
+146 -34
View File
@@ -11,14 +11,9 @@ import shutil
import sqlite3
import subprocess
import tempfile
import sys
from datetime import datetime, timezone
from pathlib import Path
# Make the top-level importers package importable when invoked as scripts/build.py.
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from importers import IMPORTERS
CVE = re.compile(r'^CVE-\d{4}-\d{4,}$', re.I)
FILES = ('vulnerabilities','aliases','affected_packages','cpe_matches')
@@ -27,6 +22,14 @@ def canonical(obj):
return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def sha(s):
return hashlib.sha256(s.encode('utf-8')).hexdigest()[:32]
def date(value):
return value if isinstance(value, str) and value else None
def get_git_revision(path):
return subprocess.check_output(['git', '-C', str(path), 'rev-parse', 'HEAD'], text=True).strip()
@@ -45,28 +48,138 @@ def input_files(repo):
yield from sorted(d.rglob('*.json'))
def stream_source(db, name, path):
importer = IMPORTERS[name]
count = 0
bad = 0
scanned = 0
def upsert_vuln(db, vid, published=None, modified=None, state=None, summary=None):
if not vid or not (CVE.match(vid) or vid.startswith(('GHSA-','OSV-','PYSEC-','RUSTSEC-','GO-','DSA-','USN-'))):
return
db.execute('''INSERT INTO vulnerabilities(id,published,modified,state,summary) VALUES(?,?,?,?,?)
ON CONFLICT(id) DO UPDATE SET
published=COALESCE(excluded.published,vulnerabilities.published),
modified=COALESCE(excluded.modified,vulnerabilities.modified),
state=COALESCE(excluded.state,vulnerabilities.state),
summary=COALESCE(excluded.summary,vulnerabilities.summary)''',
(vid, published, modified, state, summary[:500] if isinstance(summary,str) else None))
def process_cve(db, obj):
if not isinstance(obj, dict):
raise ValueError(
f"Unexpected CVE JSON type: {type(obj).__name__}; "
f"sample={str(obj)[:300]}"
)
m = obj.get('cveMetadata')
if not isinstance(m,dict):
return False
vid=m.get('cveId')
if not isinstance(vid,str) or not CVE.match(vid):
return False
upsert_vuln(db,vid,m.get('datePublished'),m.get('dateUpdated'),m.get('state'))
return True
def walk_cpe(node, allow_simple=True):
"""Only emit independent leaf matches for simple non-negated OR nodes.
Complex AND/NOT expressions must remain unsupported rather than cause false positives.
"""
if not isinstance(node,dict) or node.get('negate'):
return
if node.get('operator','OR').upper() != 'OR':
return
matches=node.get('cpeMatch') or []
if isinstance(matches,list):
for match in matches:
if isinstance(match,dict):
yield match
for child in node.get('children') or []:
yield from walk_cpe(child)
def process_nvd(db,obj):
# NVD mirror stores the CVE object directly; API responses wrap it in {'cve': ...}.
if not isinstance(obj,dict):
return False
cve=obj.get('cve') if isinstance(obj.get('cve'),dict) else obj
vid=cve.get('id')
if not isinstance(vid,str) or not CVE.match(vid):
return False
upsert_vuln(db,vid,cve.get('published'),cve.get('lastModified'),cve.get('vulnStatus'))
db.execute('DELETE FROM cpe_matches WHERE vulnerability_id=? AND source=?',(vid,'vuln-nvd'))
for node in cve.get('configurations') or []:
# NVD root configurations may contain nodes; a configuration with several
# nodes can be an AND and is not safe to flatten.
if not isinstance(node,dict) or node.get('operator','OR').upper() != 'OR' or node.get('negate'):
continue
for n in node.get('nodes') or []:
for match in walk_cpe(n):
criteria=match.get('criteria')
if not isinstance(criteria,str):
continue
vals=(criteria, int(bool(match.get('vulnerable',False))),
match.get('versionStartIncluding'), match.get('versionStartExcluding'),
match.get('versionEndIncluding'),match.get('versionEndExcluding'))
key=sha(canonical((vid,vals)))
db.execute('''INSERT OR REPLACE INTO cpe_matches
(id,vulnerability_id,criteria,vulnerable,version_start_including,version_start_excluding,version_end_including,version_end_excluding,source)
VALUES(?,?,?,?,?,?,?,?,?)''',(key,vid,*vals,'vuln-nvd'))
return True
def process_osv(db,obj):
vid=obj.get('id')
if not isinstance(vid,str) or not vid:
return False
# Only capture OSV ids with a known prefix or valid CVE.
if not (CVE.match(vid) or vid.startswith(('GHSA-','OSV-','PYSEC-','RUSTSEC-','GO-','DSA-','USN-'))):
return False
upsert_vuln(db,vid,date(obj.get('published')),date(obj.get('modified')),
'WITHDRAWN' if obj.get('withdrawn') else None,obj.get('summary'))
db.execute('DELETE FROM affected_packages WHERE vulnerability_id=? AND source=?',(vid,'vuln-osv'))
db.execute('DELETE FROM aliases WHERE vulnerability_id=?',(vid,))
for a in obj.get('aliases') or []:
if isinstance(a,str) and a:
db.execute('INSERT OR IGNORE INTO aliases(alias,vulnerability_id) VALUES(?,?)',(a,vid))
for i,a in enumerate(obj.get('affected') or []):
if not isinstance(a,dict):
continue
pkg=a.get('package') or {}
if not isinstance(pkg,dict):
continue
eco=pkg.get('ecosystem'); name=pkg.get('name')
ranges=a.get('ranges') or []
versions=a.get('versions') or []
if not isinstance(ranges,list) or not isinstance(versions,list):
continue
if not isinstance(eco,str) or not isinstance(name,str):
continue
# Preserve version/range event semantics; no lexicographic comparisons.
key=sha(canonical((vid,i,eco,name,ranges,versions)))
db.execute('INSERT OR REPLACE INTO affected_packages VALUES(?,?,?,?,?,?,?)',
(key,vid,eco,name,canonical(ranges),canonical(versions),'vuln-osv'))
return True
def stream_source(db,name,path):
count=0; bad=0
for file in input_files(path):
scanned += 1
try:
with file.open('r', encoding='utf-8') as stream:
obj = json.load(stream)
count += int(bool(importer(db, obj)))
except (ValueError, UnicodeError, TypeError, KeyError, sqlite3.Error) as exc:
bad += 1
if bad <= 5:
print(f'INVALID {name} {file}: {exc}', flush=True)
with file.open('r',encoding='utf-8') as f:
obj=json.load(f)
if name=='vuln-cve':
accepted=process_cve(db,obj)
elif name=='vuln-nvd':
accepted=process_nvd(db,obj)
elif name=='vuln-osv':
accepted=process_osv(db,obj)
else:
accepted=False
count+=int(accepted)
except (ValueError,UnicodeError,TypeError,sqlite3.Error) as exc:
bad+=1
if bad<=5:
print(f'INVALID {name} {file}: {exc}',flush=True)
if bad:
raise RuntimeError(f'{name}: {bad} invalid records; database not published')
if name == 'vuln-nvd' and count == 0:
raise RuntimeError('vuln-nvd: no usable documents; check source data and JSON shape')
if name in ('vuln-cve', 'vuln-osv') and count == 0:
raise RuntimeError(f'{name}: no usable documents; refusing to publish')
print(f'{name}: scanned={scanned} accepted={count}', flush=True)
return count
@@ -100,17 +213,17 @@ def compress(src,out):
def build_signature(source_revs, schema_path):
"""Hash source revisions plus ALL executable importer logic and schema."""
root = Path(__file__).resolve().parents[1]
"""Fingerprint inputs that can change the generated data, not wall-clock time."""
scripts_dir = Path(__file__).resolve().parent
h = hashlib.sha256()
inputs = sorted((root / 'scripts').glob('*.py'))
inputs += sorted((root / 'importers').glob('*.py'))
inputs += [schema_path.resolve()]
# Include every Python importer/helper alongside the builder, as well as SQL schema.
inputs = sorted(scripts_dir.glob('*.py'))
inputs.append(schema_path.resolve())
for path in inputs:
if not path.is_file():
raise RuntimeError(f'Missing build input: {path}')
logical_name = path.relative_to(root).as_posix() if path.is_relative_to(root) else 'sql/schema.sql'
h.update(logical_name.encode('utf-8') + b'\0')
logical_name = 'schema.sql' if path == schema_path.resolve() else f'scripts/{path.name}'
h.update(logical_name.encode('utf-8') + b'\x00')
h.update(hashlib.sha256(path.read_bytes()).digest())
h.update(canonical(source_revs).encode('utf-8'))
return h.hexdigest()
@@ -140,7 +253,7 @@ def main():
if not active:
raise SystemExit('No source repositories checked out; nothing published')
# Fail closed if expected core sources are missing.
for name in IMPORTERS:
for name in ('vuln-cve','vuln-nvd','vuln-osv'):
if name not in [n for n,_ in active]:
raise SystemExit(f'Missing core source: {name}')
source_revs={name:get_git_revision(path) for name,path in active}
@@ -165,13 +278,13 @@ def main():
db.executescript(args.schema.read_text(encoding='utf-8'))
stats={}
for name,path in active:
if name in IMPORTERS:
if name in ('vuln-cve','vuln-nvd','vuln-osv'):
stats[name]=stream_source(db,name,path)
else:
stats[name]=0
db.execute('INSERT INTO sources VALUES(?,?,?,?)',(name,source_revs[name],int(name in IMPORTERS),stats[name]))
db.execute('INSERT INTO sources VALUES(?,?,?,?)',(name,source_revs[name],int(name in ('vuln-cve','vuln-nvd','vuln-osv')),stats[name]))
print(f'{name}: indexed={stats[name]} revision={source_revs[name][:12]}',flush=True)
if not any(stats.get(n) for n in IMPORTERS):
if not any(stats.get(n) for n in ('vuln-cve','vuln-nvd','vuln-osv')):
raise RuntimeError('Core source indexes are empty; refusing to publish')
old_version=None
if old_db.is_file():
@@ -187,8 +300,7 @@ def main():
manifest={'version':version,'schema_version':1,'previous':old_version,'source_commits':source_revs,
'build_signature':signature,
'indexed_documents':stats,'format':'sqlite+zstd','delta_format':'jsonl+zstd',
'note':'CVE/NVD/OSV v1 index; AND/NOT CPE conditions are not evaluated as independent matches',
'active_importers':sorted(IMPORTERS)}
'note':'Only CVE/NVD/OSV indexed in v1; no complete version-matching engine'}
compress(new,tmp/'master.db.zst')
if old_version:
with sqlite3.connect(new) as nd:
+142 -9
View File
@@ -1,12 +1,145 @@
#!/usr/bin/env bash
# Run on the existing host-mode Gitea runner.
set -Eeuo pipefail
umask 077
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
CACHE_ROOT="${CACHE_ROOT:-/opt/gitea-builder/cache/vuln-builder}"
export CACHE_ROOT
mkdir -p "$CACHE_ROOT" "$CACHE_ROOT/releases"
exec 9>"$CACHE_ROOT/.builder.lock"
flock -w 600 9 || { echo 'Another builder is running' >&2; exit 1; }
python3 "$ROOT/scripts/sync_sources.py"
python3 "$ROOT/scripts/build.py" --sources "$CACHE_ROOT/sources" --output "$CACHE_ROOT/releases" "$@"
# ==================================================
# Configuration
# ==================================================
: "${CACHE_ROOT:=/vuln-cache/vuln-builder}"
SOURCES_DIR="${SOURCES_DIR:-$CACHE_ROOT/sources}"
SOURCES_CONFIG="${SOURCES_CONFIG:-$CACHE_ROOT/sources.json}"
SOURCE_MANIFEST="$CACHE_ROOT/source-manifest.json"
RELEASE_DIR="$CACHE_ROOT/releases"
# ==================================================
# Create directories
# ==================================================
mkdir -p "$CACHE_ROOT" "$SOURCES_DIR" "$RELEASE_DIR"
# ==================================================
# Prevent concurrent builds
# ==================================================
exec 9>"$CACHE_ROOT/.build.lock"
/usr/bin/flock -w 1800 9 || {
echo "ERROR: Another builder run is active" >&2
exit 1
}
# ==================================================
# Validate configuration
# ==================================================
echo "========================================"
echo "Vulnerability SQLite Builder"
echo "========================================"
if [ ! -s "$SOURCES_CONFIG" ]; then
echo "ERROR: sources.json missing: $SOURCES_CONFIG" >&2
exit 1
fi
if [ ! -s "$SOURCE_MANIFEST" ]; then
echo "ERROR: Source manifest missing" >&2
exit 1
fi
if [ ! -d "$SOURCES_DIR" ]; then
echo "ERROR: Source directory missing" >&2
exit 1
fi
# Verify that all sources listed in the manifest
# exist as local Git repositories.
python3 - "$SOURCE_MANIFEST" "$SOURCES_DIR" <<'PY'
import json
import subprocess
import sys
from pathlib import Path
manifest_path = Path(sys.argv[1])
sources_dir = Path(sys.argv[2])
entries = json.loads(
manifest_path.read_text(encoding="utf-8")
)
if not isinstance(entries, list) or not entries:
raise SystemExit("ERROR: Empty source manifest")
for entry in entries:
name = entry["repo"]
expected = entry["commit"]
repo = sources_dir / name
if not (repo / ".git").is_dir():
raise SystemExit(f"ERROR: Missing repository {name}")
actual = subprocess.check_output(
["git", "-C", str(repo), "rev-parse", "HEAD"],
text=True
).strip()
if actual != expected:
raise SystemExit(
f"ERROR: Commit mismatch in {name}: "
f"{actual} != {expected}"
)
print(f"OK {name}: {actual[:12]}", flush=True)
print(f"Validated {len(entries)} repositories")
PY
# ==================================================
# Build SQLite
# ==================================================
cd "$(dirname "$0")/.."
echo "Starting SQLite build..."
python3 scripts/build.py \
--sources "$SOURCES_DIR" \
--output "$RELEASE_DIR"
# ==================================================
# Verify release files
# ==================================================
echo "Checking build results..."
if [ ! -s "$RELEASE_DIR/master.db.zst" ]; then
echo "ERROR: master.db.zst missing or empty" >&2
exit 1
fi
if [ ! -s "$RELEASE_DIR/manifest.json" ]; then
echo "ERROR: Release manifest missing" >&2
exit 1
fi
# ==================================================
# Show results
# ==================================================
echo "========================================"
echo "Build completed"
echo "========================================"
ls -lh \
"$RELEASE_DIR/master.db.zst" \
"$RELEASE_DIR/manifest.json"
find "$RELEASE_DIR" \
-maxdepth 1 \
-name 'delta-*.zst' \
-printf '%f %s bytes\n'
echo "SQLite build finished successfully"
+26 -6
View File
@@ -5,13 +5,33 @@ CREATE TABLE IF NOT EXISTS vulnerabilities (id TEXT PRIMARY KEY, published TEXT,
CREATE TABLE IF NOT EXISTS aliases (alias TEXT NOT NULL, vulnerability_id TEXT NOT NULL REFERENCES vulnerabilities(id) ON DELETE CASCADE, PRIMARY KEY(alias,vulnerability_id));
CREATE INDEX IF NOT EXISTS idx_aliases_vuln ON aliases(vulnerability_id);
CREATE TABLE IF NOT EXISTS affected_packages (
id TEXT PRIMARY KEY, vulnerability_id TEXT NOT NULL REFERENCES vulnerabilities(id) ON DELETE CASCADE,
ecosystem TEXT, package_name TEXT, ranges_json TEXT NOT NULL, versions_json TEXT NOT NULL, source TEXT NOT NULL);
id TEXT PRIMARY KEY, vulnerability_id TEXT NOT NULL REFERENCES vulnerabilities(id) ON DELETE CASCADE,
ecosystem TEXT, package_name TEXT, ranges_json TEXT NOT NULL, versions_json TEXT NOT NULL, source TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS idx_affected_vuln ON affected_packages(vulnerability_id);
CREATE INDEX IF NOT EXISTS idx_affected_package ON affected_packages(ecosystem,package_name);
CREATE TABLE IF NOT EXISTS cpe_matches (
id TEXT PRIMARY KEY, vulnerability_id TEXT NOT NULL REFERENCES vulnerabilities(id) ON DELETE CASCADE,
criteria TEXT, vulnerable INTEGER, version_start_including TEXT, version_start_excluding TEXT,
version_end_including TEXT, version_end_excluding TEXT, source TEXT NOT NULL);
id TEXT PRIMARY KEY, vulnerability_id TEXT NOT NULL REFERENCES vulnerabilities(id) ON DELETE CASCADE,
criteria TEXT, vulnerable INTEGER, version_start_including TEXT, version_start_excluding TEXT,
version_end_including TEXT, version_end_excluding TEXT, source TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS idx_cpe_vuln ON cpe_matches(vulnerability_id);
-- Complex NVD AND/NOT configurations are intentionally not flattened into matches.
CREATE TABLE IF NOT EXISTS cpe_nodes (
id TEXT PRIMARY KEY,
vulnerability_id TEXT NOT NULL
REFERENCES vulnerabilities(id)
ON DELETE CASCADE,
parent_id TEXT REFERENCES cpe_nodes(id),
operator TEXT NOT NULL
CHECK (operator IN ('AND', 'OR')),
negate INTEGER NOT NULL DEFAULT 0
CHECK (negate IN (0, 1)),
source TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_cpe_nodes_vuln
ON cpe_nodes(vulnerability_id);
ALTER TABLE cpe_matches
ADD COLUMN node_id TEXT REFERENCES cpe_nodes(id);
CREATE INDEX IF NOT EXISTS idx_cpe_matches_node
ON cpe_matches(node_id);
-- Conditions expressed as AND, OR and negate are deliberately NOT converted into independent vulnerability assertions.