Get Linux kernel CVEs and its CWE

count_cwe.py

#!/usr/bin/env python

import json
import sys

def main(jsonfile):
    with open(jsonfile) as f:
        all_data = json.loads(f.read())

    cve_data = {}
    num_cves = 0
    num_no_cwe = 0
    
    for cves in all_data:
        for data in cves["vulnerabilities"]:
            cve = data["cve"]
            cveid = cve["id"]
            cwe_data = []
            num_cves += 1

            if not "weaknesses" in cve:
                continue

            weaknesses = cve["weaknesses"]
            for w in weaknesses:
                if "description" in w:
                    for desc in w["description"]:
                        v = desc["value"]
                        if v.startswith("CWE-"):
                            if not v in cwe_data:
                                cwe_data.append(v)

            if len(cwe_data) == 0:
                num_no_cwe += 1

    print(f"Total CVE: {num_cves}\nNo CWE count: {num_no_cwe}")


if __name__ == "__main__":
    if not len(sys.argv) == 2:
        print(f"[*]usage {sys.argv[0]} <cve json>")
        exit(1)

    main(sys.argv[1])

get_linux_cves.py

#!/usr/bin/env python3

import datetime
import urllib.request
import urllib.parse
import gzip
import http
import time
import json
import logging
import argparse
import os
import pprint

NVDCVE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
NVDCVE_API_KEY = None

logging.basicConfig(level = logging.INFO, format='%(asctime)s:%(levelname)s: %(message)s')
logger = logging.getLogger("linux-cve")

def nvd_request_next(url, api_key, args):
    """
    Request next part of the NVD dabase
    """
    
    request = urllib.request.Request(url + "?" + urllib.parse.urlencode(args))
    logger.debug(f"request: {request.get_full_url()}")

    if api_key:
        request.add_header("apiKey", api_key)
    logger.debug(f"Requesting {request.full_url}")

    for attempt in range(5):
        try:
            r = urllib.request.urlopen(request)

            if (r.headers['content-encoding'] == 'gzip'):
                buf = r.read()
                raw_data = gzip.decompress(buf)
            else:
                raw_data = r.read().decode("utf-8")

            r.close()
        except Exception as e:
            logger.debug(f"CVE database: received error ({e}), retrying")
            time.sleep(6)
            pass
        else:
            return raw_data
    else:
        # We failed at all attempts
        return None

def fetch_all_cves(api_key):

    index = 0
    url = NVDCVE_URL

    req_args = {}

    ret = []

    # Recommended by NVD
    sleep_time = 6
    if api_key:
        sleep_time = 2

    while True:
        logger.debug("Updating entries")

        req_args["startIndex"] = index
        req_args["cpeName"] = "cpe:2.3:o:linux:linux_kernel:-:*:*:*:*:*:*:*"

        raw_data = nvd_request_next(url, api_key, req_args)
        if raw_data is None:
            return False

        data = json.loads(raw_data)
        ret.append(data)

        index = data["startIndex"]
        total = data["totalResults"]
        per_page = data["resultsPerPage"]
        logger.debug(f"Got {per_page} entries")

        index += per_page
        if index >= total:
            break

        time.sleep(sleep_time)

    return ret

def main(args):
    cves = fetch_all_cves(args.nvd_api_key)
    if not cves:
        return False
    
    with open("linux_cves.json", "w") as f:
        json.dump(cves, f, ensure_ascii=False, indent=4)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()

    parser.add_argument("--nvd-api-key", help="API key for NVD API",
            metavar="NVDAPIKEY")
    parser.add_argument("--verbose", help="Show verbose log",
        action="store_true")
    
    args = parser.parse_args()

    if args.verbose:
        logger.setLevel(logging.DEBUG)
    
    ret = main(args)
    if not ret:
        exit(1)
    exit(0)

parse-cve-json.py

#!/usr/bin/env python

import json
import sys
import pprint

def parse(cves):
    cve_data = {}

    for data in cves["vulnerabilities"]:
        cve = data["cve"]
        cveid = cve["id"]
        cwe_data = []

        if not "weaknesses" in cve:
            continue

        weaknesses = cve["weaknesses"]
        for w in weaknesses:
            if "description" in w:
                for desc in w["description"]:
                    v = desc["value"]
                    if v.startswith("CWE-"):
                        if not v in cwe_data:
                            cwe_data.append(v)

        cve_data[cveid] = cwe_data

    return cve_data

def main(jsonfile):
    with open(jsonfile) as f:
        cves = json.loads(f.read())

    summary = {}
    cve_data = {}
    for data in cves:
        tmp = parse(data)
        cve_data.update(tmp)

    summary = {}
    for cve in cve_data:
        cwes = cve_data[cve]
        for cwe in cwes:
            if not cwe in summary:
                summary[cwe] = 1
            else:
                summary[cwe] += 1
    
    summary = reversed(sorted(summary.items(), key=lambda item: item[1]))

    with open("linux_cve_cwe_report.csv", "w") as f:
        f.write("CWE, Count\n")
        for d in summary:
            f.write(f"{d[0]}, {d[1]}\n")

    with open("linux_cve_cwe_data.csv", "w") as f:
        f.write("CVE, CWE\n")
        for cveid in cve_data:
            cwes = cve_data[cveid]
            for cwe in cwes:
                f.write(f"{cveid}, {cwe}\n")

if __name__ == "__main__":
    if not len(sys.argv) == 2:
        print(f"[*]usage {sys.argv[0]} <cve json>")
        exit(1)

    main(sys.argv[1])