import argparse import datetime import glob import json import os import sys import time from collections import defaultdict from dataclasses import dataclass from urllib.parse import quote import requests from bs4 import BeautifulSoup from dataclass_wizard import JSONSerializable from semantic_version import Version, SimpleSpec @dataclass class Vulnerability: id: str level: str description: str versions: list[str] @dataclass class NpmPackage: name: str vulnerabilities: list[Vulnerability] class Snyk: BASE_URL = "https://security.snyk.io" HEADERS = { "User-Agent": "IT Security Research" } @staticmethod def scrape_npm_package(name: str): resp = requests.get(f"{Snyk.BASE_URL}/package/npm/{quote(name, safe='')}", headers=Snyk.HEADERS) resp.raise_for_status() vulnerabilities = [] try: doc = BeautifulSoup(resp.text, 'html.parser') table = doc.find(True, {"class": "vulns-table"}) rows = table.find("tbody").find_all("tr") for row in rows: vulnerabilities.append(Vulnerability( id=row.a["href"].rsplit("/", 1)[-1], level=row.abbr["title"], description=row.a.text.strip(), versions=[v for c in row.find_all(True, {"class": "vulnerable-versions__chip"}) for v in c.text.strip().split()], )) except AttributeError: pass return NpmPackage(name, vulnerabilities) class GithubAdvisories: BASE_URL = "https://api.github.com/advisories" HEADERS = { "User-Agent": "IT Security Research", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", } LAST_RESPONSE_HEADERS = {} @staticmethod def scrape_npm_package(name: str): headers = GithubAdvisories.HEADERS.copy() if os.getenv("GITHUB_TOKEN") is not None: headers["Authorization"] = f"Bearer {os.getenv('GITHUB_TOKEN')}" if int(GithubAdvisories.LAST_RESPONSE_HEADERS.get("x-ratelimit-remaining", "1")) == 0: reset_time = int(GithubAdvisories.LAST_RESPONSE_HEADERS.get("x-ratelimit-reset", "0")) print(f"GithubAdvisories: Sleeping until rate limit is available again at {datetime.datetime.fromtimestamp(reset_time)}") time.sleep(time.time() - reset_time) resp = requests.get(f"{GithubAdvisories.BASE_URL}", params={"affects": name}, headers=headers) time.sleep(0.1) GithubAdvisories.LAST_RESPONSE_HEADERS = resp.headers resp.raise_for_status() data = json.loads(resp.text) vulnerabilities = [] for vuln in data: vulnerabilities.append(Vulnerability( id=vuln["ghsa_id"], level=vuln["severity"], description=vuln["summary"], versions=[s2.strip() for s in vuln["vulnerabilities"] for s2 in s["vulnerable_version_range"].split(",")], )) return NpmPackage(name, vulnerabilities) class NpmRegistry: BASE_URL = "https://registry.npmjs.com" HEADERS = { "User-Agent": "IT Security Research", } @staticmethod def fetch_package_meta(name: str): try: resp = requests.get(f"{NpmRegistry.BASE_URL}/{name}", headers=NpmRegistry.HEADERS) resp.raise_for_status() return resp.json()["time"] except (requests.HTTPError, json.JSONDecodeError, AttributeError): return {} @dataclass class DatabaseEntry(JSONSerializable): source: str pkg: NpmPackage class VulnerabilityDatabase: def __init__(self, database_file: str = None): self.pkglist = set() self.database = defaultdict(list) self.releases: dict[str, dict[str, str]] = {} if database_file is not None: try: with open(database_file, "rb") as f: data = json.load(f) for k, v in data[0].items(): self.database[k] = DatabaseEntry.from_list(v) self.releases = data[1] except (FileNotFoundError, json.decoder.JSONDecodeError) as e: print(f"Failed to load existing database", e, file=sys.stderr) self.pkglist.update(self.database.keys()) def fetch_missing(self, snyk=True, github_adv=True): for pkg in self.pkglist: if pkg not in self.releases: self.releases[pkg] = NpmRegistry.fetch_package_meta(pkg) if pkg in self.releases and len(self.releases[pkg]) > 0 and "unpublished" not in self.releases[pkg]: # don't search package if npm doesn't have a record if snyk and len([e for e in self.database[pkg] if e.source == "SNYK"]) == 0: self.database[pkg].append(DatabaseEntry(source="SNYK", pkg=Snyk.scrape_npm_package(pkg))) if github_adv and len([e for e in self.database[pkg] if e.source == "GITHUB_ADV"]) == 0: self.database[pkg].append(DatabaseEntry(source="GITHUB_ADV", pkg=GithubAdvisories.scrape_npm_package(pkg))) def save(self, database_file: str): with open(database_file, "w") as f: json.dump([self.database, self.releases], f, default=lambda o: o.to_dict()) def is_vulnerable(self, pkg: str, version: str) -> bool: version = Version(version) for advisory in self.database[pkg]: for vuln in advisory.pkg.vulnerabilities: if len(vuln.versions) > 0 and all([SimpleSpec(spec).match(version) for spec in vuln.versions]): return True return False def get_vulnerabilities(self, pkg: str, version: str) -> list[Vulnerability]: vulnerabilities = [] version = Version(version) for advisory in self.database[pkg]: for vuln in advisory.pkg.vulnerabilities: if all([SimpleSpec(spec).match(version) for spec in vuln.versions]): vulnerabilities.append(vuln) return vulnerabilities if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument("database_file", type=str) args = parser.parse_args() index_dir = os.getenv("INDEX_DIR", ".") indexed_pkgs = set() for i in glob.iglob("*.index.json", root_dir=index_dir): indexed_pkgs.add(i[:-11].replace("+", "/", 1)) db = VulnerabilityDatabase(args.database_file) print(f"Fetching {len(indexed_pkgs.difference(db.pkglist))} package data") db.pkglist.update(indexed_pkgs) try: db.fetch_missing(github_adv=False) finally: db.save(args.database_file)