34 lines
1.6 KiB
Python
34 lines
1.6 KiB
Python
"""OSV advisories, retaining complete version range semantics as JSON."""
|
|
from .common import canonical, is_vulnerability_id, sha, upsert_vuln, valid_date
|
|
|
|
|
|
def process(db, obj):
|
|
if not isinstance(obj, dict):
|
|
return False
|
|
vid = obj.get('id')
|
|
if not is_vulnerability_id(vid):
|
|
return False
|
|
upsert_vuln(db, vid, valid_date(obj.get('published')), valid_date(obj.get('modified')),
|
|
'WITHDRAWN' if obj.get('withdrawn') else None, obj.get('summary'))
|
|
db.execute('DELETE FROM affected_packages WHERE vulnerability_id=? AND source=?', (vid, 'vuln-osv'))
|
|
db.execute('DELETE FROM aliases WHERE vulnerability_id=?', (vid,))
|
|
for alias in obj.get('aliases') or []:
|
|
if isinstance(alias, str) and alias:
|
|
db.execute('INSERT OR IGNORE INTO aliases(alias,vulnerability_id) VALUES(?,?)', (alias, vid))
|
|
for i, item in enumerate(obj.get('affected') or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
package = item.get('package') or {}
|
|
if not isinstance(package, dict):
|
|
continue
|
|
eco, name = package.get('ecosystem'), package.get('name')
|
|
ranges, versions = item.get('ranges') or [], item.get('versions') or []
|
|
if not isinstance(eco, str) or not isinstance(name, str):
|
|
continue
|
|
if not isinstance(ranges, list) or not isinstance(versions, list):
|
|
continue
|
|
key = sha(canonical((vid, i, eco, name, ranges, versions)))
|
|
db.execute('INSERT OR REPLACE INTO affected_packages VALUES(?,?,?,?,?,?,?)',
|
|
(key, vid, eco, name, canonical(ranges), canonical(versions), 'vuln-osv'))
|
|
return True
|