vuln-suse/scripts/sync.py

#!/usr/bin/env python3
"""Mirror SUSE/openSUSE CSAF advisory documents using the official index."""
import argparse
import concurrent.futures
import hashlib
import json
import os
from pathlib import Path
import re
import sys
import time
from datetime import datetime, timezone
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

BASE = os.getenv('SUSE_CSAF_URL', 'https://ftp2.gwdg.de/pub/linux/suse/projects/security/csaf/').rstrip('/') + '/'
NAME = re.compile(r'^(?:opensuse|suse)-[a-z0-9_-]*?(20\d\d)[_-][a-z0-9_.-]+\.json$', re.I)
# Handles 'suse-su-2026_...' etc; do not interpret CVE year as advisory year.
USER_AGENT='apex-project-vuln-suse-mirror/1.0'

def fetch(url, tries=5):
    for i in range(tries):
        try:
            with urlopen(Request(url, headers={'User-Agent': USER_AGENT, 'Accept':'application/json,text/plain,*/*'}), timeout=90) as r:
                return r.read()
        except (HTTPError, URLError, TimeoutError) as e:
            if i == tries-1 or (isinstance(e, HTTPError) and e.code not in (429,500,502,503,504)):
                raise
            time.sleep(min(60, 2**(i+1)))

def eligible(name, year):
    if '/' in name or '\\' in name or '..' in name or not name.endswith('.json'):
        return False
    m = NAME.match(name)
    return bool(m and int(m.group(1)) >= year)

def load_index(data, year):
    text=data.decode('utf-8-sig')
    names=set()
    for line in text.splitlines():
        name=line.strip()
        if eligible(name,year): names.add(name)
    if not names:
        raise ValueError('Index liefert keine passenden CSAF-Dateien; kein Abgleich durchgeführt')
    return names

def valid_doc(blob):
    doc=json.loads(blob)
    if not isinstance(doc,dict) or not isinstance(doc.get('document'),dict):
        raise ValueError('Ungültiges CSAF-Dokument (document fehlt)')
    if not isinstance(doc.get('product_tree'),dict) and not isinstance(doc.get('vulnerabilities'),list):
        raise ValueError('CSAF enthält weder product_tree noch vulnerabilities')
    return doc

def atomic_write(path,blob):
    path.parent.mkdir(parents=True,exist_ok=True)
    tmp=path.with_name(path.name+'.tmp')
    try:
        tmp.write_bytes(blob)
        tmp.replace(path)
    finally:
        tmp.unlink(missing_ok=True)

def main(argv=None):
    ap=argparse.ArgumentParser()
    ap.add_argument('--since-year',type=int,default=2017)
    ap.add_argument('--workers',type=int,default=3)
    ap.add_argument('--review',type=int,default=300,help='Anzahl älterer Advisories pro Lauf erneut herunterladen')
    ap.add_argument('--full',action='store_true',help='Alle Dokumente erneut abgleichen')
    ap.add_argument('--max-files',type=int,default=0,help='Nur für Tests, keine Löschungen')
    args=ap.parse_args(argv)
    root=Path(__file__).resolve().parent.parent
    out=root/'data'/'csaf'
    state_file=root/'state'/'sync.json'
    state=json.loads(state_file.read_text()) if state_file.exists() else {}
    index=fetch(BASE+'index.txt')
    names=load_index(index,args.since_year)
    previous=set(state.get('index',[]))
    known={p.name for p in out.glob('*/*.json')}
    missing=names-known
    remaining=sorted(names-missing)
    pos=int(state.get('review_cursor',0))
    review=[]
    if args.full:
        targets=sorted(names)
    else:
        if remaining:
            count=min(args.review,len(remaining))
            review=[remaining[(pos+i)%len(remaining)] for i in range(count)]
        targets=sorted(missing|set(review))
    if args.max_files:
        targets=targets[:args.max_files]
    print(f'Index={len(names)} vorhanden={len(known)} neu={len(missing)} Abrufe={len(targets)}',flush=True)
    errors=[]; changed=0
    def worker(name):
        m=NAME.match(name)
        path=out/m.group(1)/name
        blob=fetch(BASE+quote(name))
        valid_doc(blob)
        if path.exists() and hashlib.sha256(path.read_bytes()).digest()==hashlib.sha256(blob).digest():
            return False
        atomic_write(path,blob)
        return True
    with concurrent.futures.ThreadPoolExecutor(max_workers=max(1,min(8,args.workers))) as pool:
        futures={pool.submit(worker,n):n for n in targets}
        for fut in concurrent.futures.as_completed(futures):
            try: changed+=int(fut.result())
            except Exception as e: errors.append(f'{futures[fut]}: {e}')
    if errors:
        for msg in errors[:30]:print('ERROR',msg,file=sys.stderr)
        raise RuntimeError(f'{len(errors)} Downloads fehlgeschlagen. Keine Löschungen und kein Status-Commit.')
    if not args.max_files:
        deleted=0
        # The public index is authoritative only after a successful complete fetch of the index and all requested documents.
        for old in sorted(known-names):
            for path in out.glob('*/'+old):
                path.unlink(); deleted+=1
        atomic_write(root/'data'/'index.txt',index)
        new_state={'source':BASE,'since_year':args.since_year,'updated_at':datetime.now(timezone.utc).isoformat(),
                   'index':sorted(names),'review_cursor':(pos+len(review)) % max(1,len(remaining)),
                   'documents':len(names)}
        atomic_write(state_file,(json.dumps(new_state,ensure_ascii=False,indent=2)+'\n').encode())
        print(f'Änderungen={changed}, Löschungen={deleted}')
    else:
        print('TESTMODUS: Status und Löschungen unverändert')
    return 0

if __name__=='__main__':
    try:sys.exit(main())
    except Exception as exc:
        print(f'FEHLER: {exc}',file=sys.stderr); sys.exit(1)