204 lines
8.6 KiB
Python
204 lines
8.6 KiB
Python
#!/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()
|
|
|