71 lines
3.1 KiB
Python
71 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Mirror official CVE List V5 current snapshot (2017+) into this repository."""
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
UPSTREAM = 'https://github.com/CVEProject/cvelistV5.git'
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
def run(*args, cwd=None):
|
|
return subprocess.check_output(args, cwd=cwd, text=True).strip()
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('--since-year', type=int, default=2017)
|
|
ap.add_argument('--source-url', default=UPSTREAM)
|
|
ap.add_argument('--minimum-records', type=int, default=1000)
|
|
args = ap.parse_args()
|
|
now_year = dt.datetime.now(dt.timezone.utc).year
|
|
if not 1999 <= args.since_year <= now_year:
|
|
ap.error('invalid since-year')
|
|
if not shutil.which('git') or not shutil.which('rsync'):
|
|
raise SystemExit('git and rsync are required')
|
|
source_dir = ROOT / 'data'
|
|
state_dir = ROOT / 'state'
|
|
source_dir.mkdir(exist_ok=True)
|
|
state_dir.mkdir(exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix='vuln-cve-') as tmp:
|
|
clone = Path(tmp) / 'source'
|
|
subprocess.run(['git', 'clone', '--depth=1', '--filter=blob:none', '--sparse', args.source_url, str(clone)], check=True)
|
|
years = [f'cves/{y}' for y in range(args.since_year, now_year + 1)]
|
|
subprocess.run(['git', '-C', str(clone), 'sparse-checkout', 'set', *years], check=True)
|
|
revision = run('git', 'rev-parse', 'HEAD', cwd=clone)
|
|
record_root = clone / 'cves'
|
|
candidates = list(record_root.rglob('CVE-*.json'))
|
|
if len(candidates) < args.minimum_records:
|
|
raise RuntimeError(f'Upstream has only {len(candidates)} CVE records: refusing destructive synchronization')
|
|
# Validate JSON structure and identifier across the entire source before deleting local entries.
|
|
validated = 0
|
|
for p in candidates:
|
|
if not re.fullmatch(r'CVE-\d{4}-\d{4,}', p.stem):
|
|
raise RuntimeError(f'Unexpected file name: {p}')
|
|
try:
|
|
with p.open(encoding='utf-8') as handle:
|
|
record = json.load(handle)
|
|
if record.get('cveMetadata', {}).get('cveId') != p.stem:
|
|
raise ValueError('metadata CVE ID mismatch')
|
|
except Exception as exc:
|
|
raise RuntimeError(f'Invalid upstream CVE record {p}: {exc}') from exc
|
|
validated += 1
|
|
# Delete files not present upstream, including records withdrawn or removed.
|
|
subprocess.run(['rsync', '-a', '--delete', '--exclude=.gitkeep', str(record_root) + '/', str(source_dir) + '/'], check=True)
|
|
state = {
|
|
'source': args.source_url,
|
|
'upstream_commit': revision,
|
|
'since_year': args.since_year,
|
|
'records': validated,
|
|
'synced_at_utc': dt.datetime.now(dt.timezone.utc).isoformat(),
|
|
}
|
|
(state_dir / 'sync.json').write_text(json.dumps(state, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
|
print(f'Mirrored {validated} CVE records at {revision}')
|
|
|
|
if __name__ == '__main__':
|
|
main()
|