Initial Debian security mirror
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user