#!/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, staging_root): """Stream bz2 archive to disk. Validate all selected documents before publishing. Returns (document_count, manifest_digest). RAM remains bounded to one member. """ import codecs staging_root.mkdir(parents=True, exist_ok=True) count = 0 manifest = hashlib.sha256() # Store just names and hashes (small) to detect conflicting duplicates. seen = {} 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}') source = tf.extractfile(member) if source is None: raise ValueError(f'Unreadable member: {member.name}') raw = source.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): raise ValueError(f'Missing CSAF document object: {member.name}') if 'vulnerabilities' in obj and not isinstance(obj['vulnerabilities'], list): raise ValueError(f'Invalid vulnerabilities field: {member.name}') year = match.group(1) cve = f'CVE-{year}-{match.group(2)}' rel = f'{year}/{cve}.json' digest = hashlib.sha256(raw).hexdigest() if rel in seen: if seen[rel] != digest: raise ValueError(f'Duplicate conflicting CVE document: {rel}') continue seen[rel] = digest atomic_write(staging_root / rel, raw) count += 1 if count % 5000 == 0: print(f'Staged and validated {count} VEX documents', flush=True) if not count: raise ValueError('Zero CSAF VEX documents; refusing snapshot') for rel in sorted(seen): manifest.update(rel.encode('utf-8') + b'\0' + seen[rel].encode('ascii') + b'\n') return count, manifest.hexdigest() def same_file(path_a, path_b): if not path_a.exists() or path_a.stat().st_size != path_b.stat().st_size: return False with path_a.open('rb') as a, path_b.open('rb') as b: while True: left = a.read(1024 * 1024) right = b.read(1024 * 1024) if left != right: return False if not left: return True def reconcile(staging_root, root, state, source, since_year, count, digest): """Apply only after entire archive has been staged and validated.""" root.mkdir(parents=True, exist_ok=True) added = changed = removed = 0 staged_paths = set() for source_path in staging_root.rglob('*.json'): rel = source_path.relative_to(staging_root) staged_paths.add(rel.as_posix()) target = root / rel if same_file(target, source_path): continue existed = target.exists() atomic_write(target, source_path.read_bytes()) if existed: changed += 1 else: added += 1 for existing in root.rglob('*.json'): if existing.relative_to(root).as_posix() not in staged_paths: existing.unlink() removed += 1 payload = {'source': source, 'min_year': since_year, 'record_count': count, 'snapshot_sha256': digest, '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 # Stage under persistent cache filesystem; always clean up after the run. with tempfile.TemporaryDirectory(prefix='staging-', dir=str(args.cache)) as staging: stage = Path(staging) count, digest = parse_archive(archive, args.since_year, stage) result = reconcile(stage, Path('data/vex'), Path('state/sync.json'), args.source, args.since_year, count, digest) print(json.dumps(result,indent=2)) if __name__=='__main__': main()