vuln-microsoft/scripts/sync.py

#!/usr/bin/env python3
"""Mirrors full MSRC CVRF JSON documents and the raw update index."""
import argparse
import datetime as dt
import json
import os
from pathlib import Path
import random
import re
import sys
import time
import urllib.error
import urllib.request

BASE = os.getenv('MSRC_BASE_URL', 'https://api.msrc.microsoft.com/cvrf/v3.0').rstrip('/')
DOC_RE = re.compile(r'^([12][0-9]{3})-([A-Za-z]{3})$')
HEADERS = {'Accept': 'application/json', 'User-Agent': 'apex-project-vuln-microsoft/1.0'}

def fetch(url, attempts=5):
    for n in range(attempts):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=HEADERS), timeout=90) as response:
                data = response.read()
            obj = json.loads(data)
            if not isinstance(obj, (dict, list)):
                raise ValueError('Unexpected response JSON type')
            return obj
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, ValueError) as exc:
            if n == attempts - 1:
                raise RuntimeError(f'Download failed: {url}: {exc}') from exc
            pause = min(60, 2 ** n * 3) + random.random()
            print(f'Retrying {url}: {exc}; {pause:.1f}s', file=sys.stderr, flush=True)
            time.sleep(pause)

def atomic_json(path, obj):
    path.parent.mkdir(parents=True, exist_ok=True)
    content = (json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(',', ':')) + '\n').encode()
    if path.exists() and path.read_bytes() == content:
        return False
    temp = path.with_suffix(path.suffix + '.tmp')
    temp.write_bytes(content)
    os.replace(temp, path)
    return True

def index_rows(index):
    if isinstance(index, list):
        rows = index
    elif isinstance(index, dict):
        rows = index.get('value') or index.get('Value')
    else:
        rows = None
    if not isinstance(rows, list) or not rows:
        raise ValueError('Microsoft update index has no entries; aborting safely')
    return rows

def normalize_id(row):
    value = row.get('ID') or row.get('Id') or row.get('id')
    if not isinstance(value, str) or not DOC_RE.fullmatch(value):
        return None
    return value

def release_date(row):
    value = row.get('CurrentReleaseDate') or row.get('currentReleaseDate') or row.get('InitialReleaseDate') or row.get('initialReleaseDate')
    if not value:
        return None
    try:
        return dt.datetime.fromisoformat(str(value).replace('Z', '+00:00')).astimezone(dt.timezone.utc)
    except ValueError:
        return None

def sync(root, since_year, days, full, delay):
    data_root = root / 'data'
    state_file = root / 'state' / 'sync.json'
    today = dt.datetime.now(dt.timezone.utc)
    index = fetch(BASE + '/updates')
    rows = index_rows(index)
    filtered = {}
    for row in rows:
        if not isinstance(row, dict):
            raise ValueError('Non-object update entry')
        docid = normalize_id(row)
        if docid and int(docid[:4]) >= since_year:
            filtered[docid] = row
    if not filtered:
        raise ValueError('No documents after year filter; aborting')
    previous = json.loads(state_file.read_text()) if state_file.exists() else {}
    prev_dates = previous.get('release_dates', {})
    selected = []
    for docid, row in sorted(filtered.items()):
        target = data_root / 'cvrf' / docid[:4] / (docid + '.json')
        date_val = row.get('CurrentReleaseDate') or row.get('currentReleaseDate') or row.get('InitialReleaseDate') or row.get('initialReleaseDate')
        date_key = str(date_val or '')
        recent = (release_date(row) or dt.datetime(1970, 1, 1, tzinfo=dt.timezone.utc)) >= today - dt.timedelta(days=days)
        if full or not target.exists() or prev_dates.get(docid) != date_key or recent:
            selected.append((docid, target))
    print(f'MSRC documents listed={len(filtered)}, download={len(selected)}, full={full}', flush=True)
    updated = 0
    for idx, (docid, target) in enumerate(selected, start=1):
        doc = fetch(BASE + '/cvrf/' + docid)
        if not isinstance(doc, dict) or not doc:
            raise ValueError(f'Empty or invalid CVRF document {docid}')
        updated += atomic_json(target, doc)
        if idx % 20 == 0 or idx == len(selected):
            print(f'processed {idx}/{len(selected)}, changed={updated}', flush=True)
        if delay and idx < len(selected):
            time.sleep(delay)
    atomic_json(data_root / 'index' / 'updates.json', index)
    atomic_json(state_file, {
        'source': BASE,
        'since_year': since_year,
        'last_success_utc': today.isoformat(),
        'documents': len(filtered),
        'processed': len(selected),
        'changed': updated,
        'release_dates': {id_: str(row.get('CurrentReleaseDate') or row.get('currentReleaseDate') or row.get('InitialReleaseDate') or row.get('initialReleaseDate') or '') for id_, row in filtered.items()}
    })
    print('SUCCESS', flush=True)

if __name__ == '__main__':
    ap = argparse.ArgumentParser()
    ap.add_argument('--root', type=Path, default=Path('.'))
    ap.add_argument('--since-year', type=int, default=2017)
    ap.add_argument('--recheck-days', type=int, default=45)
    ap.add_argument('--full', action='store_true')
    ap.add_argument('--delay', type=float, default=0.5)
    args = ap.parse_args()
    sync(args.root.resolve(), args.since_year, args.recheck_days, args.full, args.delay)