Initial NVD mirror

2026-11-10 18:41:04 +01:00
commit ab460f1d97
8 changed files with 341 additions and 0 deletions
@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""NVD CVE API 2.0 mirror. Requires Python >=3.10, no third-party modules."""
+import argparse
+import json
+import os
+import random
+import re
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+API = 'https://services.nvd.nist.gov/rest/json/cves/2.0'
+ROOT = Path(__file__).resolve().parents[1]
+CVE_RE = re.compile(r'^CVE-(\d{4})-\d{4,}$')
+UTC = timezone.utc
+
+def timestamp(dt):
+    return dt.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%S.000+00:00')
+
+def parse_time(s):
+    return datetime.fromisoformat(s.replace('Z', '+00:00')).astimezone(UTC)
+
+def atomic_json(path, obj):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = (json.dumps(obj, ensure_ascii=False, separators=(',', ':'), sort_keys=True) + '\n').encode('utf-8')
+    tmp = path.with_name(path.name + '.tmp')
+    with tmp.open('wb') as f:
+        f.write(content)
+    os.replace(tmp, path)
+
+def windows(start, end, days=30):
+    while start < end:
+        nxt = min(start + timedelta(days=days), end)
+        yield start, nxt
+        start = nxt
+
+class Client:
+    def __init__(self, api_key=None, delay=None, retries=6):
+        self.api_key = api_key
+        self.delay = delay if delay is not None else (0.8 if api_key else 6.5)
+        self.retries = retries
+        self.last_call = 0
+
+    def fetch(self, params):
+        query = urllib.parse.urlencode(params)
+        url = API + '?' + query
+        headers = {'User-Agent': 'vuln-nvd-mirror/1.0 (+https://git.apex-project.de/VULN/vuln-nvd)', 'Accept': 'application/json'}
+        if self.api_key:
+            headers['apiKey'] = self.api_key
+        for attempt in range(self.retries + 1):
+            wait = self.delay - (time.monotonic() - self.last_call)
+            if wait > 0:
+                time.sleep(wait)
+            self.last_call = time.monotonic()
+            try:
+                with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=120) as resp:
+                    data = json.load(resp)
+                if not isinstance(data, dict) or not isinstance(data.get('vulnerabilities'), list) or not isinstance(data.get('totalResults'), int):
+                    raise ValueError('Unexpected NVD API JSON payload')
+                return data
+            except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as exc:
+                status = getattr(exc, 'code', None)
+                if status is not None and status not in (403, 408, 429, 500, 502, 503, 504):
+                    raise
+                if attempt >= self.retries:
+                    raise
+                retry_after = None
+                if isinstance(exc, urllib.error.HTTPError):
+                    retry_after = exc.headers.get('Retry-After')
+                try:
+                    backoff = min(300, max(float(retry_after), 1)) if retry_after else min(240, 10 * 2**attempt + random.random() * 4)
+                except ValueError:
+                    backoff = min(240, 10 * 2**attempt)
+                print(f'API error {exc}; retry {attempt+1}/{self.retries} after {backoff:.0f}s', file=sys.stderr, flush=True)
+                time.sleep(backoff)
+
+def import_period(client, start, end, date_prefix, min_year, counts):
+    params = {date_prefix+'StartDate': timestamp(start), date_prefix+'EndDate': timestamp(end), 'resultsPerPage': 2000}
+    offset = 0
+    while True:
+        response = client.fetch({**params, 'startIndex': offset})
+        records = response['vulnerabilities']
+        total = response['totalResults']
+        if not records and offset < total:
+            raise RuntimeError(f'Empty page before totalResults at offset {offset}/{total}')
+        for item in records:
+            if not isinstance(item, dict) or not isinstance(item.get('cve'), dict):
+                raise ValueError('Malformed NVD vulnerability')
+            cve = item['cve']
+            identifier = cve.get('id', '')
+            match = CVE_RE.fullmatch(identifier)
+            if not match:
+                raise ValueError(f'Invalid CVE ID: {identifier!r}')
+            if int(match.group(1)) < min_year:
+                counts['skipped'] += 1
+                continue
+            destination = ROOT / 'data' / match.group(1) / f'{identifier}.json'
+            payload = (json.dumps(cve, sort_keys=True, ensure_ascii=False, separators=(',', ':')) + '\n').encode('utf-8')
+            if destination.exists() and destination.read_bytes() == payload:
+                counts['unchanged'] += 1
+            else:
+                atomic_json(destination, cve)
+                counts['written'] += 1
+        counts['seen'] += len(records)
+        offset += len(records)
+        print(f'  {date_prefix}: {start.date()} to {end.date()}, records {offset}/{total}', flush=True)
+        if offset >= total:
+            break
+    return counts
+
+def run(args):
+    if args.since_year < 1999:
+        raise ValueError('--since-year must be >= 1999')
+    statefile = ROOT / 'state' / 'sync.json'
+    state = json.loads(statefile.read_text(encoding='utf-8')) if statefile.exists() else {}
+    end = parse_time(args.end) if args.end else datetime.now(UTC) - timedelta(minutes=5)
+    end = end.replace(microsecond=0)
+    client = Client(api_key=os.getenv('NVD_API_KEY'), delay=args.delay)
+    mode = args.mode if args.mode != 'auto' else ('incremental' if state.get('initial_complete') else 'full')
+    if mode == 'incremental' and not state.get('initial_complete'):
+        raise ValueError('Initial full import required before incremental mode')
+    start = datetime(args.since_year, 1, 1, tzinfo=UTC) if mode == 'full' else parse_time(state['last_successful_end']) - timedelta(hours=args.overlap_hours)
+    if start >= end:
+        print('Already up to date')
+        return
+    counts = {'seen': 0, 'written': 0, 'unchanged': 0, 'skipped': 0}
+    prefix = 'pub' if mode == 'full' else 'lastMod'
+    print(f'Mode={mode} from={timestamp(start)} to={timestamp(end)} with API key={bool(client.api_key)}', flush=True)
+    for a, b in windows(start, end, args.window_days):
+        import_period(client, a, b, prefix, args.since_year, counts)
+    atomic_json(statefile, {
+        'schema_version': 1,
+        'source': 'NVD CVE API 2.0',
+        'since_year': args.since_year,
+        'initial_complete': True,
+        'last_successful_end': end.isoformat(),
+        'last_run_mode': mode,
+        'last_run_counts': counts,
+        'updated_at': datetime.now(UTC).isoformat()
+    })
+    print('Success:', json.dumps(counts), flush=True)
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--since-year', type=int, default=2017)
+    parser.add_argument('--mode', choices=('auto', 'full', 'incremental'), default='auto')
+    parser.add_argument('--overlap-hours', type=int, default=48)
+    parser.add_argument('--window-days', type=int, default=30)
+    parser.add_argument('--delay', type=float, default=None)
+    parser.add_argument('--end', help='Optional UTC ISO timestamp (testing/backfill)')
+    args = parser.parse_args()
+    if not 1 <= args.window_days <= 119 or args.overlap_hours < 0:
+        parser.error('window-days must be 1..119; overlap-hours must be nonnegative')
+    try:
+        run(args)
+    except Exception as exc:
+        print(f'FAILED, sync state not advanced: {exc}', file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == '__main__':
+    main()
@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""Validate JSON files and CVE identity; nonzero exit status on errors."""
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+ROOT = Path(__file__).resolve().parents[1]
+PAT = re.compile(r'CVE-(\d{4})-\d{4,}')
+
+def main():
+    p = argparse.ArgumentParser()
+    p.add_argument('--changed-only', action='store_true', help='Validate files modified in the last Git worktree diff (including staged)')
+    args = p.parse_args()
+    if args.changed_only:
+        import subprocess
+        changed = subprocess.run(['git', 'status', '--porcelain', '-uall'], cwd=ROOT, capture_output=True, text=True, check=True).stdout
+        files = []
+        for line in changed.splitlines():
+            if len(line) < 4:
+                continue
+            path = line[3:]
+            if path.startswith('data/') and path.endswith('.json') and (ROOT / path).is_file():
+                files.append(ROOT / path)
+    else:
+        files = list((ROOT / 'data').glob('*/*.json'))
+    errors = []
+    for path in files:
+        try:
+            record = json.loads(path.read_text(encoding='utf-8'))
+            identifier = record['id']
+            match = PAT.fullmatch(identifier)
+            assert match and path.name == identifier + '.json' and path.parent.name == match.group(1)
+            assert isinstance(record.get('published'), str)
+            assert isinstance(record.get('lastModified'), str)
+            assert isinstance(record.get('vulnStatus'), str)
+        except Exception as exc:
+            errors.append(f'{path}: {exc}')
+    for error in errors[:30]:
+        print(error, file=sys.stderr)
+    print(f'Validated {len(files)} files; errors: {len(errors)}')
+    sys.exit(1 if errors else 0)
+
+if __name__ == '__main__':
+    main()