259 lines
8.4 KiB
Python
259 lines
8.4 KiB
Python
|
"""
|
||
|
|
Autarch CVE Database — CVE lookup for Android packages via OSV.dev.
|
||
|
|
by: SsSnake
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import threading
|
||
|
|
import urllib.request
|
||
|
|
import urllib.error
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
_db_instance = None
|
||
|
|
_db_lock = threading.Lock()
|
||
|
|
|
||
|
|
|
||
|
|
def get_cve_db():
|
||
|
|
global _db_instance
|
||
|
|
if _db_instance is None:
|
||
|
|
with _db_lock:
|
||
|
|
if _db_instance is None:
|
||
|
|
_db_instance = CveDatabase()
|
||
|
|
return _db_instance
|
||
|
|
|
||
|
|
|
||
|
|
class CveDatabase:
|
||
|
|
"""CVE lookup for Android packages. Uses OSV.dev as primary data source."""
|
||
|
|
|
||
|
|
DATA_DIR = Path(__file__).parent.parent / 'data' / 'ioc' / 'cve'
|
||
|
|
OSV_QUERY_URL = 'https://api.osv.dev/v1/query'
|
||
|
|
OSV_BATCH_URL = 'https://api.osv.dev/v1/querybatch'
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
|
self._cache = {} # package_name -> [vuln_dicts]
|
||
|
|
self._stats_cache = None
|
||
|
|
self._lock = threading.Lock()
|
||
|
|
self._last_query = None
|
||
|
|
|
||
|
|
def get_cve_for_package(self, package_name, version=None) -> list:
|
||
|
|
"""Query OSV.dev for CVEs affecting an Android package."""
|
||
|
|
with self._lock:
|
||
|
|
if package_name in self._cache:
|
||
|
|
return self._cache[package_name]
|
||
|
|
|
||
|
|
try:
|
||
|
|
body = {
|
||
|
|
"package": {
|
||
|
|
"name": package_name,
|
||
|
|
"ecosystem": "Android"
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if version is not None:
|
||
|
|
body["version"] = version
|
||
|
|
|
||
|
|
response = self._query_osv(body)
|
||
|
|
vulns = response.get("vulns", [])
|
||
|
|
|
||
|
|
results = []
|
||
|
|
for vuln in vulns:
|
||
|
|
parsed = self._parse_vuln(vuln, package_name)
|
||
|
|
results.append(parsed)
|
||
|
|
|
||
|
|
with self._lock:
|
||
|
|
self._cache[package_name] = results
|
||
|
|
self._last_query = datetime.now(timezone.utc).isoformat()
|
||
|
|
|
||
|
|
return results
|
||
|
|
except Exception:
|
||
|
|
return []
|
||
|
|
|
||
|
|
def get_cve_for_android_sdk(self, sdk_version) -> list:
|
||
|
|
"""Query OSV.dev for CVEs affecting a specific Android SDK version."""
|
||
|
|
package_name = f"android:{sdk_version}"
|
||
|
|
try:
|
||
|
|
body = {
|
||
|
|
"package": {
|
||
|
|
"name": package_name,
|
||
|
|
"ecosystem": "Android"
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
response = self._query_osv(body)
|
||
|
|
vulns = response.get("vulns", [])
|
||
|
|
|
||
|
|
results = []
|
||
|
|
for vuln in vulns:
|
||
|
|
parsed = self._parse_vuln(vuln, package_name)
|
||
|
|
results.append(parsed)
|
||
|
|
|
||
|
|
with self._lock:
|
||
|
|
self._cache[package_name] = results
|
||
|
|
self._last_query = datetime.now(timezone.utc).isoformat()
|
||
|
|
|
||
|
|
return results
|
||
|
|
except Exception:
|
||
|
|
return []
|
||
|
|
|
||
|
|
def check_packages_batch(self, packages: list) -> dict:
|
||
|
|
"""Query OSV.dev for CVEs affecting multiple Android packages at once."""
|
||
|
|
if not packages:
|
||
|
|
return {}
|
||
|
|
|
||
|
|
try:
|
||
|
|
results = {}
|
||
|
|
# Split into chunks of 50
|
||
|
|
chunk_size = 50
|
||
|
|
for i in range(0, len(packages), chunk_size):
|
||
|
|
chunk = packages[i:i + chunk_size]
|
||
|
|
queries = [
|
||
|
|
{"package": {"name": p["name"], "ecosystem": "Android"}}
|
||
|
|
for p in chunk
|
||
|
|
]
|
||
|
|
body = {"queries": queries}
|
||
|
|
|
||
|
|
data = json.dumps(body).encode("utf-8")
|
||
|
|
req = urllib.request.Request(
|
||
|
|
self.OSV_BATCH_URL,
|
||
|
|
data=data,
|
||
|
|
headers={
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
"User-Agent": "Autarch/1.0"
|
||
|
|
},
|
||
|
|
method="POST"
|
||
|
|
)
|
||
|
|
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
|
|
response = json.loads(resp.read().decode("utf-8"))
|
||
|
|
except Exception:
|
||
|
|
continue
|
||
|
|
|
||
|
|
batch_results = response.get("results", [])
|
||
|
|
for idx, result in enumerate(batch_results):
|
||
|
|
if idx >= len(chunk):
|
||
|
|
break
|
||
|
|
pkg_name = chunk[idx]["name"]
|
||
|
|
vulns = result.get("vulns", [])
|
||
|
|
parsed_vulns = [self._parse_vuln(v, pkg_name) for v in vulns]
|
||
|
|
results[pkg_name] = parsed_vulns
|
||
|
|
|
||
|
|
with self._lock:
|
||
|
|
self._cache[pkg_name] = parsed_vulns
|
||
|
|
|
||
|
|
with self._lock:
|
||
|
|
if results:
|
||
|
|
self._last_query = datetime.now(timezone.utc).isoformat()
|
||
|
|
|
||
|
|
return results
|
||
|
|
except Exception:
|
||
|
|
return {}
|
||
|
|
|
||
|
|
def get_stats(self) -> dict:
|
||
|
|
"""Return basic stats about the CVE database cache."""
|
||
|
|
with self._lock:
|
||
|
|
return {
|
||
|
|
"total_cached": len(self._cache),
|
||
|
|
"last_query": self._last_query,
|
||
|
|
"data_dir": str(self.DATA_DIR)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _query_osv(self, body: dict) -> dict:
|
||
|
|
"""POST a JSON body to the OSV.dev query endpoint. Returns parsed JSON or {}."""
|
||
|
|
try:
|
||
|
|
data = json.dumps(body).encode("utf-8")
|
||
|
|
req = urllib.request.Request(
|
||
|
|
self.OSV_QUERY_URL,
|
||
|
|
data=data,
|
||
|
|
headers={
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
"User-Agent": "Autarch/1.0"
|
||
|
|
},
|
||
|
|
method="POST"
|
||
|
|
)
|
||
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
|
|
return json.loads(resp.read().decode("utf-8"))
|
||
|
|
except Exception:
|
||
|
|
return {}
|
||
|
|
|
||
|
|
def _parse_vuln(self, vuln: dict, package_name: str) -> dict:
|
||
|
|
"""Extract standardised CVE fields from an OSV vulnerability object."""
|
||
|
|
# cve_id: prefer alias starting with "CVE-", else use vuln id
|
||
|
|
cve_id = vuln.get("id", "UNKNOWN")
|
||
|
|
for alias in vuln.get("aliases", []):
|
||
|
|
if alias.startswith("CVE-"):
|
||
|
|
cve_id = alias
|
||
|
|
break
|
||
|
|
|
||
|
|
# severity: try database_specific first, then severity array score, normalise
|
||
|
|
raw_severity = (
|
||
|
|
vuln.get("database_specific", {}).get("severity")
|
||
|
|
or vuln.get("severity", [{}])[0].get("score", "UNKNOWN")
|
||
|
|
)
|
||
|
|
severity = self._normalise_severity(raw_severity)
|
||
|
|
|
||
|
|
# description: summary or details, capped at 500 chars
|
||
|
|
description = vuln.get("summary", vuln.get("details", ""))[:500]
|
||
|
|
|
||
|
|
# patched_version: look for a "fixed" event in affected ranges
|
||
|
|
patched_version = "unknown"
|
||
|
|
affected_list = vuln.get("affected", [{}])
|
||
|
|
if affected_list:
|
||
|
|
ranges = affected_list[0].get("ranges", [])
|
||
|
|
for r in ranges:
|
||
|
|
for event in r.get("events", []):
|
||
|
|
if "fixed" in event:
|
||
|
|
patched_version = event["fixed"]
|
||
|
|
break
|
||
|
|
if patched_version != "unknown":
|
||
|
|
break
|
||
|
|
|
||
|
|
return {
|
||
|
|
"package": package_name,
|
||
|
|
"cve_id": cve_id,
|
||
|
|
"severity": severity,
|
||
|
|
"description": description,
|
||
|
|
"patched_version": patched_version,
|
||
|
|
"source": "osv.dev"
|
||
|
|
}
|
||
|
|
|
||
|
|
def _normalise_severity(self, raw) -> str:
|
||
|
|
"""Map a raw severity string or CVSS score to critical/high/medium/low."""
|
||
|
|
if raw is None or raw == "UNKNOWN":
|
||
|
|
return "medium"
|
||
|
|
|
||
|
|
raw_str = str(raw).strip().upper()
|
||
|
|
|
||
|
|
# Direct label match
|
||
|
|
label_map = {
|
||
|
|
"CRITICAL": "critical",
|
||
|
|
"HIGH": "high",
|
||
|
|
"MEDIUM": "medium",
|
||
|
|
"MODERATE": "medium",
|
||
|
|
"LOW": "low",
|
||
|
|
"NONE": "low"
|
||
|
|
}
|
||
|
|
if raw_str in label_map:
|
||
|
|
return label_map[raw_str]
|
||
|
|
|
||
|
|
# Try numeric CVSS score
|
||
|
|
try:
|
||
|
|
score = float(raw_str)
|
||
|
|
if score >= 9.0:
|
||
|
|
return "critical"
|
||
|
|
elif score >= 7.0:
|
||
|
|
return "high"
|
||
|
|
elif score >= 4.0:
|
||
|
|
return "medium"
|
||
|
|
else:
|
||
|
|
return "low"
|
||
|
|
except ValueError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# CVSS vector strings often start with digits after the last colon
|
||
|
|
# e.g. "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
|
||
|
|
# Fall back to medium if we can't parse it
|
||
|
|
return "medium"
|