diff --git a/scripts/sync.py b/scripts/sync.py index 372d3283..1af61e04 100644 --- a/scripts/sync.py +++ b/scripts/sync.py @@ -80,66 +80,99 @@ def fetch(url, archive, meta_path, timeout, retries): 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={} +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) + 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: + 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) + 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}') - # CSAF Base allows vulnerabilities to be omitted. A strict VEX-profile - # check would require it, but the upstream archive also contains - # document-only files. Preserve raw upstream files without claiming - # those files individually conform to the VEX profile. 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)}' - # 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 + 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 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()) +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)) @@ -157,9 +190,14 @@ def main(): 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) + # 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() +