scripts/sync.py aktualisiert
This commit is contained in:
+83
-45
@@ -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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user