61 lines
2.6 KiB
Python
61 lines
2.6 KiB
Python
"""NVD 2.0 direct CVE JSON and wrapped API responses.
|
|
|
|
Only independent, non-negated, OR-only configurations can be flattened into
|
|
cpe_matches. Complex AND / NOT conditions are skipped to avoid false positives.
|
|
"""
|
|
from .common import CVE, canonical, sha, upsert_vuln
|
|
|
|
|
|
def simple_or_matches(config):
|
|
if not isinstance(config, dict) or config.get('negate'):
|
|
return
|
|
if str(config.get('operator', 'OR')).upper() != 'OR':
|
|
return
|
|
nodes = config.get('nodes') or []
|
|
if not isinstance(nodes, list):
|
|
return
|
|
# Skip the WHOLE configuration if any subtree contains an AND or a negate.
|
|
def safe(node):
|
|
return (isinstance(node, dict) and not node.get('negate')
|
|
and str(node.get('operator', 'OR')).upper() == 'OR'
|
|
and isinstance(node.get('cpeMatch') or [], list)
|
|
and isinstance(node.get('children') or [], list)
|
|
and all(safe(child) for child in node.get('children') or []))
|
|
if not all(safe(node) for node in nodes):
|
|
return
|
|
def walk(node):
|
|
for match in node.get('cpeMatch') or []:
|
|
if isinstance(match, dict):
|
|
yield match
|
|
for child in node.get('children') or []:
|
|
yield from walk(child)
|
|
for node in nodes:
|
|
yield from walk(node)
|
|
|
|
|
|
def process(db, obj):
|
|
if not isinstance(obj, dict):
|
|
return False
|
|
cve = obj.get('cve') if isinstance(obj.get('cve'), dict) else obj
|
|
vid = cve.get('id')
|
|
if not isinstance(vid, str) or not CVE.fullmatch(vid):
|
|
return False
|
|
upsert_vuln(db, vid, cve.get('published'), cve.get('lastModified'), cve.get('vulnStatus'))
|
|
db.execute('DELETE FROM cpe_matches WHERE vulnerability_id=? AND source=?', (vid, 'vuln-nvd'))
|
|
configs = cve.get('configurations') or []
|
|
if not isinstance(configs, list):
|
|
raise ValueError(f'Invalid configurations for {vid}')
|
|
for config in configs:
|
|
for match in simple_or_matches(config):
|
|
criteria = match.get('criteria')
|
|
if not isinstance(criteria, str):
|
|
continue
|
|
vals = (criteria, int(bool(match.get('vulnerable', False))),
|
|
match.get('versionStartIncluding'), match.get('versionStartExcluding'),
|
|
match.get('versionEndIncluding'), match.get('versionEndExcluding'))
|
|
key = sha(canonical((vid, vals)))
|
|
db.execute('''INSERT OR REPLACE INTO cpe_matches
|
|
(id,vulnerability_id,criteria,vulnerable,version_start_including,version_start_excluding,version_end_including,version_end_excluding,source)
|
|
VALUES(?,?,?,?,?,?,?,?,?)''', (key, vid, *vals, 'vuln-nvd'))
|
|
return True
|