70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Mirror the official Debian Security Tracker JSON export without changing its contents."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError
|
|
|
|
URL = 'https://security-tracker.debian.org/tracker/data/json'
|
|
|
|
def sync(root: Path, url: str, timeout: int = 240):
|
|
data = root / 'data' / 'tracker.json'
|
|
state = root / 'state' / 'sync.json'
|
|
meta = {}
|
|
if state.exists():
|
|
meta = json.loads(state.read_text(encoding='utf-8'))
|
|
headers = {'User-Agent': 'apex-project-vuln-debian-mirror/1.0', 'Accept': 'application/json'}
|
|
if data.exists():
|
|
if meta.get('etag'): headers['If-None-Match'] = meta['etag']
|
|
if meta.get('last_modified'): headers['If-Modified-Since'] = meta['last_modified']
|
|
try:
|
|
with urlopen(Request(url, headers=headers), timeout=timeout) as response:
|
|
raw = response.read()
|
|
etag = response.headers.get('ETag')
|
|
modified = response.headers.get('Last-Modified')
|
|
except HTTPError as exc:
|
|
if exc.code == 304 and data.exists():
|
|
print('HTTP 304: keine Änderungen am Debian-Export')
|
|
return False
|
|
raise
|
|
if len(raw) < 100: raise RuntimeError('Antwort ist unerwartet kurz')
|
|
parsed = json.loads(raw)
|
|
validate(parsed)
|
|
digest = hashlib.sha256(raw).hexdigest()
|
|
if data.exists() and hashlib.sha256(data.read_bytes()).hexdigest() == digest:
|
|
print('Inhalt unverändert; kein Commit erforderlich')
|
|
return False
|
|
data.parent.mkdir(parents=True, exist_ok=True)
|
|
state.parent.mkdir(parents=True, exist_ok=True)
|
|
temp = data.with_suffix('.json.tmp')
|
|
temp.write_bytes(raw)
|
|
os.replace(temp, data)
|
|
state_temp = state.with_suffix('.json.tmp')
|
|
state_temp.write_text(json.dumps({'source': url, 'sha256': digest, 'etag':etag,
|
|
'last_modified':modified, 'downloaded_at':datetime.now(timezone.utc).isoformat(),
|
|
'package_count':len(parsed)}, indent=2, ensure_ascii=False)+'\n', encoding='utf-8')
|
|
os.replace(state_temp, state)
|
|
print(f'Debian Export gespeichert: {len(raw):,} Bytes / {len(parsed):,} Pakete')
|
|
return True
|
|
|
|
def validate(parsed):
|
|
if not isinstance(parsed, dict) or not parsed:
|
|
raise ValueError('Erwartet nichtleeres JSON-Objekt mit Paketnamen als Keys')
|
|
for package, cves in list(parsed.items())[:200]:
|
|
if not isinstance(package, str) or not isinstance(cves, dict):
|
|
raise ValueError(f'Unerwartetes Debian-JSON-Schema bei {package!r}')
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument('--root', type=Path, default=Path(__file__).resolve().parent.parent)
|
|
p.add_argument('--url', default=URL)
|
|
p.add_argument('--timeout', type=int, default=240)
|
|
args = p.parse_args()
|
|
sync(args.root, args.url, args.timeout)
|
|
|
|
if __name__ == '__main__': main()
|