Initial Rocky Linux security mirror

2026-11-11 21:06:27 +01:00
commit c8424e92e9
10 changed files with 235 additions and 0 deletions
@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+import json, os, re, time
+from datetime import datetime,timezone
+from pathlib import Path
+from urllib.parse import urlencode, quote
+from common import request,store,state_file
+
+ROOT=Path(__file__).resolve().parents[1]
+API=os.getenv('ROCKY_API','https://apollo.build.resf.org/v2/advisories').rstrip('/')
+PAGE_SIZE=int(os.getenv('ROCKY_PAGE_SIZE','100'))
+MAX_PAGES=int(os.getenv('ROCKY_MAX_PAGES','20'))
+DELAY=float(os.getenv('ROCKY_DELAY','0.15'))
+STATE=ROOT/'state/sync.json'
+
+def get_advisories(body):
+    obj=json.loads(body)
+    if not isinstance(obj,dict) or not isinstance(obj.get('advisories'),list):
+        raise ValueError('Unexpected Apollo list response')
+    total=int(obj['total']) if 'total' in obj else None
+    if total is None: raise ValueError('Apollo response lacks total count')
+    return obj['advisories'],total
+
+def identity(doc):
+    name=doc.get('name') or doc.get('id')
+    if not isinstance(name,str) or not re.fullmatch(r'[A-Za-z0-9:._-]+',name):
+        raise ValueError(f'Invalid advisory identity {name!r}')
+    return name
+
+def date_filter(doc):
+    return identity(doc).startswith(('RLSA','RXSA'))
+
+def main():
+    old=json.loads(STATE.read_text()) if STATE.exists() else {}
+    page=max(1,int(old.get('next_page',1)))
+    pages=0
+    while pages<MAX_PAGES:
+        q=urlencode({'page':page,'limit':PAGE_SIZE,'filters.type':'TYPE_SECURITY','filters.includeRpms':'false'})
+        _,body,_=request(API+'?'+q)
+        entries,total=get_advisories(body)
+        if not entries:
+            if page==1: raise RuntimeError('Apollo returned zero advisories')
+            page=1
+            break
+        count=0
+        for summary in entries:
+            if not date_filter(summary): continue
+            name=identity(summary)
+            # Individual API responses retain affected RPM mappings. Do not use only list summaries.
+            url=API+'/'+quote(name,safe='')
+            _,detail,_=request(url)
+            parsed=json.loads(detail)
+            advisory=parsed.get('advisory') if isinstance(parsed,dict) else None
+            if not isinstance(advisory,dict) or identity(advisory)!=name:
+                raise ValueError(f'Invalid detail for {name}')
+            year=re.search(r'(20\d\d)',name)
+            if not year: raise ValueError(f'No year for {name}')
+            dest=ROOT/'data/advisories'/year.group(1)/(name.replace(':','-')+'.json')
+            store(dest,detail)
+            count+=1
+            time.sleep(DELAY)
+        print(f'Rocky page {page}: {len(entries)} indexed, {count} details, total={total}',flush=True)
+        page+=1; pages+=1
+        if page>(total+PAGE_SIZE-1)//PAGE_SIZE: page=1
+        state_file(STATE,{'next_page':page,'last_success_utc':datetime.now(timezone.utc).isoformat(),'last_total':total})
+        if page==1: break
+
+if __name__=='__main__': main()