130 lines
5.8 KiB
Python
130 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Mirror public Oracle security advisories (HTML and linked CSAF JSON)."""
|
|
import argparse
|
|
import hashlib
|
|
from html.parser import HTMLParser
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
from urllib.parse import urljoin, urlparse
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError, URLError
|
|
|
|
BASE = 'https://www.oracle.com'
|
|
INDEX = BASE + '/security-alerts/'
|
|
ARCHIVE = BASE + '/security-alerts/cpuarchive.html'
|
|
UA = 'vuln-oracle-mirror/1.0 (+security data mirror; contact site administrator)'
|
|
YEAR = re.compile(r'\b(20\d{2})\b')
|
|
|
|
class Links(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.links=[]; self.href=None; self.parts=[]
|
|
def handle_starttag(self, tag, attrs):
|
|
if tag=='a':
|
|
self.href=dict(attrs).get('href'); self.parts=[]
|
|
def handle_data(self, data):
|
|
if self.href is not None: self.parts.append(data)
|
|
def handle_endtag(self,tag):
|
|
if tag=='a' and self.href is not None:
|
|
self.links.append((self.href,' '.join(''.join(self.parts).split())))
|
|
self.href=None; self.parts=[]
|
|
|
|
def get(url, tries=4):
|
|
last=None
|
|
for i in range(tries):
|
|
try:
|
|
with urlopen(Request(url, headers={'User-Agent':UA,'Accept':'text/html,application/json,*/*'}), timeout=90) as response:
|
|
body=response.read(30_000_001)
|
|
if len(body)>30_000_000: raise ValueError('Document too large')
|
|
return body
|
|
except (HTTPError, URLError, TimeoutError) as exc:
|
|
last=exc
|
|
if isinstance(exc,HTTPError) and exc.code in (400,401,403,404): break
|
|
time.sleep(min(2**i,12))
|
|
raise RuntimeError(f'Failed to fetch {url}: {last}')
|
|
|
|
def parse_links(data,url):
|
|
p=Links(); p.feed(data.decode('utf-8','replace'))
|
|
return [(urljoin(url,href),label) for href,label in p.links]
|
|
|
|
def allowed(url):
|
|
return urlparse(url).hostname in ('www.oracle.com','oracle.com') and url.startswith('https://')
|
|
|
|
def discover(index_html, archive_html, since):
|
|
found={}
|
|
for page,src in ((index_html,INDEX),(archive_html,ARCHIVE)):
|
|
for url,label in parse_links(page,src):
|
|
if not allowed(url): continue
|
|
path=urlparse(url).path.lower()
|
|
if not path.startswith('/security-alerts/') or not path.endswith('.html'): continue
|
|
if path.endswith(('cpuarchive.html','index.html')): continue
|
|
title=label.lower()
|
|
if not any(x in title for x in ('critical patch update','critical security patch update','security alert','solaris third party bulletin')): continue
|
|
# Years are from the advisory title, not CVE numbers or page modification date.
|
|
years=[int(y) for y in YEAR.findall(label)]
|
|
if not years or max(years)<since: continue
|
|
found[url]={'url':url,'title':label,'year':max(years)}
|
|
return sorted(found.values(),key=lambda v:v['url'])
|
|
|
|
def csaf_links(html,url):
|
|
result=[]
|
|
for link,label in parse_links(html,url):
|
|
if not allowed(link): continue
|
|
path=urlparse(link).path.lower()
|
|
if path.endswith('.json') and ('csaf' in label.lower() or 'csaf' in path):
|
|
result.append(link)
|
|
return sorted(set(result))
|
|
|
|
def write_changed(path,contents):
|
|
path.parent.mkdir(parents=True,exist_ok=True)
|
|
if path.exists() and path.read_bytes()==contents: return False
|
|
temp=path.with_name(path.name+'.tmp')
|
|
temp.write_bytes(contents)
|
|
os.replace(temp,path)
|
|
return True
|
|
|
|
def run(root,since,delay):
|
|
root=Path(root)
|
|
index=get(INDEX); archive=get(ARCHIVE)
|
|
items=discover(index,archive,since)
|
|
if len(items)<25:
|
|
raise RuntimeError(f'Only {len(items)} advisories discovered: page structure may have changed; refusing incomplete sync')
|
|
changed=0; errors=[]; csaf_count=0
|
|
changed+=write_changed(root/'data/index/security-alerts.html',index)
|
|
changed+=write_changed(root/'data/index/cpuarchive.html',archive)
|
|
for i,item in enumerate(items,1):
|
|
url=item['url']; filename=urlparse(url).path.rsplit('/',1)[-1]
|
|
target=root/'data/advisories'/str(item['year'])/filename
|
|
try:
|
|
html=get(url)
|
|
if b'<html' not in html.lower() and b'<!doctype html' not in html.lower():
|
|
raise ValueError('Unexpected non-HTML response')
|
|
links=csaf_links(html,url)
|
|
# Fetch JSON before committing HTML, fail full run if a linked CSAF is unreachable.
|
|
for csaf in links:
|
|
raw=get(csaf); doc=json.loads(raw)
|
|
if not isinstance(doc,dict): raise ValueError('CSAF must be object')
|
|
filename_json=urlparse(csaf).path.rsplit('/',1)[-1]
|
|
csaf_path=root/'data/csaf'/str(item['year'])/filename_json
|
|
changed+=write_changed(csaf_path,raw)
|
|
csaf_count+=1
|
|
changed+=write_changed(target,html)
|
|
print(f'[{i}/{len(items)}] {filename}: {len(links)} CSAF',flush=True)
|
|
except Exception as e:
|
|
errors.append(f'{url}: {e}')
|
|
print(f'ERROR {url}: {e}',flush=True)
|
|
if delay: time.sleep(delay)
|
|
if errors:
|
|
raise RuntimeError(f'{len(errors)} advisory errors; sync state not updated. First: {errors[0]}')
|
|
state={'source':INDEX,'since_year':since,'advisory_count':len(items),'csaf_links_found':csaf_count,
|
|
'advisories':items}
|
|
changed+=write_changed(root/'state/catalog.json',(json.dumps(state,ensure_ascii=False,indent=2)+'\n').encode())
|
|
print(f'Complete: {len(items)} advisories, {csaf_count} CSAF links, {changed} changed files')
|
|
|
|
if __name__=='__main__':
|
|
p=argparse.ArgumentParser(); p.add_argument('--root',default='.'); p.add_argument('--since-year',type=int,default=2017); p.add_argument('--delay',type=float,default=0.2)
|
|
a=p.parse_args(); run(a.root,a.since_year,a.delay)
|