#!/usr/bin/env python3 """SUSE CSAF VEX mirror from official upstream tarball. Atomic snapshot: download complete archive, validate every selected JSON before changing the dataset, then reconcile data, including deletions. Resume: existing uncommitted files are retained and compared against the next valid snapshot. """ import argparse import hashlib import json import os from pathlib import Path import re import shutil import sys import tarfile import tempfile import time from datetime import datetime, timezone from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen DEFAULT_URL = 'https://ftp.suse.com/pub/projects/security/csaf-vex.tar.bz2' CVE_RE = re.compile(r'(?i)cve[-_](\d{4})[-_](\d{4,})') MAX_MEMBER = 32 * 1024 * 1024 UA = 'vuln-suse-vex-mirror/1.0 (+https://git.apex-project.de/VULN/vuln-suse-vex)' def atomic_write(path, data): path.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix='.tmp-') try: with os.fdopen(fd, 'wb') as f: f.write(data) f.flush() os.fsync(f.fileno()) os.replace(tmp, path) finally: if os.path.exists(tmp): os.unlink(tmp) def fetch(url, archive, meta_path, timeout, retries): meta = json.loads(meta_path.read_text()) if meta_path.exists() else {} headers = {'User-Agent': UA} if archive.exists() and archive.stat().st_size > 0 and meta.get('source') == url: if meta.get('etag'): headers['If-None-Match'] = meta['etag'] if meta.get('last_modified'): headers['If-Modified-Since'] = meta['last_modified'] for n in range(retries): try: req = Request(url, headers=headers) with urlopen(req, timeout=timeout) as response: fd, tmp = tempfile.mkstemp(dir=str(archive.parent), prefix='.download-') try: with os.fdopen(fd, 'wb') as f: shutil.copyfileobj(response, f, length=1024*1024) f.flush(); os.fsync(f.fileno()) if os.stat(tmp).st_size < 1024: raise ValueError('Archive unexpectedly small') # Validate compressed tar before replacing persistent archive with tarfile.open(tmp, 'r:bz2') as tf: if tf.next() is None: raise ValueError('Empty upstream tar') os.replace(tmp, archive) new_meta = {'source':url, 'etag':response.headers.get('ETag'), 'last_modified':response.headers.get('Last-Modified')} atomic_write(meta_path,json.dumps(new_meta, indent=2).encode()) return True finally: if os.path.exists(tmp):os.unlink(tmp) except HTTPError as e: if e.code == 304 and archive.exists():return False if e.code not in (429, 500, 502, 503, 504):raise err=e except (OSError, URLError, ValueError, tarfile.TarError) as e: err=e if n+1 < retries: delay=min(2**n * 8, 120) print(f'Download failed: {err}; retry in {delay}s',file=sys.stderr,flush=True) time.sleep(delay) raise RuntimeError(f'Cannot fetch {url}: {err}') def parse_archive(path, since_year): """Read and validate ALL relevant raw VEX JSON before any output mutation.""" items={} with tarfile.open(path, 'r|bz2') as tf: for member in tf: if not member.isfile() or not member.name.lower().endswith('.json'): continue name=Path(member.name).name match=CVE_RE.search(name) if not match or int(match.group(1)) < since_year: continue if member.size <= 0 or member.size > MAX_MEMBER: raise ValueError(f'Invalid JSON member size: {member.name}') f=tf.extractfile(member) if not f:raise ValueError(f'Unreadable member: {member.name}') raw=f.read(MAX_MEMBER+1) if len(raw) != member.size or len(raw)>MAX_MEMBER: raise ValueError(f'Incomplete JSON: {member.name}') obj=json.loads(raw) if not isinstance(obj,dict) or not isinstance(obj.get('document'),dict) or not isinstance(obj.get('vulnerabilities'),list): raise ValueError(f'Invalid CSAF document: {member.name}') year=match.group(1) cve=f'CVE-{year}-{match.group(2)}' # Duplicate names across archive: only exact duplicate bytes are acceptable target=f'{year}/{cve}.json' if target in items and items[target] != raw: raise ValueError(f'Duplicate conflicting CVE document: {target}') items[target]=raw if not items:raise ValueError('Zero CSAF VEX documents; refusing snapshot') return items def reconcile(items, root, state, source, since_year): root.mkdir(parents=True,exist_ok=True) existing={p.relative_to(root).as_posix():p for p in root.rglob('*.json')} added=changed=removed=0 for rel,raw in items.items(): target=root/rel if target.exists() and target.read_bytes()==raw:continue added += not target.exists(); changed += target.exists() atomic_write(target,raw) for rel,p in existing.items(): # Files older than since_year never belong to this mirror if rel not in items: p.unlink();removed+=1 # This state describes a completely validated snapshot, not an incomplete run payload={'source':source,'min_year':since_year,'record_count':len(items), 'snapshot_sha256':hashlib.sha256('\n'.join(sorted(items)).encode()).hexdigest(), 'completed_at':datetime.now(timezone.utc).isoformat(), 'added':added,'updated':changed,'deleted':removed} atomic_write(state,json.dumps(payload,indent=2,sort_keys=True).encode()) return payload def main(): ap=argparse.ArgumentParser() ap.add_argument('--source',default=os.environ.get('SUSE_VEX_URL',DEFAULT_URL)) ap.add_argument('--since-year',type=int,default=2017) ap.add_argument('--cache',type=Path,default=Path(os.environ.get('SUSE_VEX_DOWNLOAD_CACHE','/var/lib/act_runner/cache/vuln-suse-vex/download'))) ap.add_argument('--timeout',type=int,default=180) ap.add_argument('--retries',type=int,default=4) args=ap.parse_args() if args.since_year<1999:ap.error('since-year must be >= 1999') args.cache.mkdir(parents=True,exist_ok=True) archive=args.cache/'csaf-vex.tar.bz2' meta=args.cache/'http.json' downloaded=fetch(args.source,archive,meta,args.timeout,args.retries) # Don't rewrite the state, and don't regenerate a commit if HTTP returns 304. if not downloaded and Path('state/sync.json').exists(): print('Upstream archive unchanged (HTTP 304)') return records=parse_archive(archive,args.since_year) result=reconcile(records,Path('data/vex'),Path('state/sync.json'),args.source,args.since_year) print(json.dumps(result,indent=2)) if __name__=='__main__': main()