68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
#!/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()
|