path: root/english/security/oval/generate.py

#!/usr/bin/python3

"""
generate.py

Extracts the data from the security tracker and creates OVAL queries to
be used with the OVAL query interpreter (see https://oval.mitre.org)

Copyright (c) 2015 Nicholas Luedtke
          (c) 2016 Sebastien Delafond <sdelafond@gmail.com>
          (c) 2023 Carsten Schoenert <c.schoenert@t-online.de>

SPDX-License-Identifier: GPL-2.0
"""

import argparse
import json
import os
import logging
import pprint
import re
import subprocess
import sys
from typing import Any

import oval.definition.generator
from oval.parser import tracker

# TODO:
# - these may need changed or reworked.
# - ideally they would be extracted from the release information @ website
DEBIAN_VERSION = {
    "potato": "2",
    "sarge": "3",
    "woody": "3",
    "etch": "4",
    "lenny": "5",
    "squeeze": "6",
    "wheezy": "7",
    "jessie": "8",
    "stretch": "9",
    "buster": "10",
    "bullseye": "11",
    "bookworm": "12",
    "trixie": "13",
    "forky": "14",
    "sid": "1000",
}


def printdsas(ovals):
    """Generate and print OVAL Definitions for collected DSA information"""
    logging.debug(pprint.pformat(ovals))
    ovalDefinitions = oval.definition.generator.createOVALDefinitions(ovals)
    oval.definition.generator.printOVALDefinitions(ovalDefinitions)


def get_key(secref, package, dsaRef=None):
    if secref.startswith("CVE") or secref.startswith("TEMP"):
        # A CVE can affect multiple source packages, so include that
        # source package name in the key as well
        key = f"{secref} {package}"
    else:
        # Either a bug number, or the dsaRef itself, and therefore
        # affecting only one source package: in this case we want to
        # generate one *single* entry with that dsaRef key, and It
        # will later be exported as a patch/advisory OVAL entity.
        key = dsaRef

    return key


def add_dsa_info(ovals, dsaResult, wmlResult, dsaRef, debian_release):
    logging.debug("working on dsa info %s", pprint.pformat(dsaResult))

    debian_version = DEBIAN_VERSION[debian_release]

    secrefs = dsaResult[1].get("secrefs", [])

    # tack on dsaRef to make sure we always create a DSA entry in the
    # ovals dictionary, even if this DSA was not linked to any bug number
    for secref in secrefs + [
        dsaRef,
    ]:
        key = get_key(secref, dsaResult[1]["packages"], dsaRef)
        logging.debug("DSA working on secref '%s', considering key '%s'", secref, key)

        if key not in ovals:
            logging.debug("new entry %s: %s", key, dsaResult[1])
            ovals[key] = dsaResult[1]

        # add info from .data file
        logging.debug("DSA enriching existing entry '%s' with '%s'", key, dsaResult[1])
        for k, v in dsaResult[1].items():
            if k not in ovals[key]:
                ovals[key][k] = v

        # skip if the wml file does not contain the debian release
        if debian_version in wmlResult[1]:
            logging.debug("add_dsa_result wmlResult[1]: %s", wmlResult[1])
            # add info from .wml file to CVE in
            ovals = add_wml_info(ovals, wmlResult, key, dsaRef, debian_release)

    return ovals


def add_wml_info(ovals, wmlResult, key, dsaRef, debian_release):
    debian_version = DEBIAN_VERSION[debian_release]
    entry = ovals[key]
    wml_data, releases = wmlResult
    logging.debug("WML enriching existing entry %s with %s", key, wml_data)

    for k, v in wml_data.items():
        if k == "moreinfo" and k not in entry:
            entry["dsa"] = dsaRef
            entry[k] = v.replace("\n", " ").strip()
        elif k == "description" and k not in entry:
            entry[k] = v
        else:
            entry[k] = v
    if "release" not in entry:
        entry["release"] = {}
    entry["release"].update({debian_version: releases[debian_version]})
    logging.debug("WML after adding wml data %s", entry["release"])

    return ovals


def collect_dsa_data(
        ovals: dict[str, Any],
        debian_release: str,
        dsa_data_file: str) -> None:
    """ Collect DSA data and kick out XML data

    Extract and collect data from the tracker file of the Security team as a
    dictionary and initate the creation of the XML data.

    Parameters:
        ovals (dict): Dictionary containing packages with CVE data.
        debian_release (str): The Debian release number e.g "12".
        dsa_data_file (str): Reference to tracker data from the Security Team.
    Returns:
        None
    """
    data = tracker.parse_tracker_data(dsa_data_file, DEBIAN_VERSION)
    for key, value in data.items():
        dsaResult = value[0]
        #print(f"\ndsaResult\n {dsaResult}")

        wmlResult = value[1], value[2]
        #print(f"\nwmlResult\n {wmlResult}")

        dsaRef = key
        #print(f"\ndsaRef\n {dsaRef}")
        if dsaResult and wmlResult:
            ovals = add_dsa_info(
                ovals, dsaResult, wmlResult, dsaRef, debian_release
            )


def parseJSON(ovals, json_data, debian_release):
    """
    Parse the JSON data and extract information needed for OVAL definitions
    :param json_data: Json_Data
    :return:
    """

    logging.debug("Start of JSON Parse.")
    for package in json_data:
        logging.debug("Parsing package %s", package)
        for cve in json_data[package]:
            logging.debug("Getting releases for %s", cve)
            release = {}
            for rel in json_data[package][cve]["releases"]:
                status = json_data[package][cve]["releases"][rel]["status"]
                f_str = "no"

                if status == "resolved":
                    fixed_v = json_data[package][cve]["releases"][rel]["fixed_version"]
                    f_str = "yes"
                else:
                    fixed_v = "0"
                    f_str = "no"

                if status == "resolved" and fixed_v == "0":
                    # This CVE never impacted the given release
                    logging.debug("Release %s not affected by %s", rel, cve)
                    continue

                if debian_release == rel:
                    release.update({DEBIAN_VERSION[rel]: {"all": {package: fixed_v}}})
                    key = get_key(cve, package)
                    ovals.update(
                        {
                            key: {
                                "packages": package,
                                "title": key,
                                "vulnerable": "yes",
                                "date": None,
                                "fixed": f_str,
                                "description": json_data[package][cve].get(
                                    "description", ""
                                ),
                                "secrefs": (cve,),
                                "release": release,
                            }
                        }
                    )
                    logging.debug("Created entry for %s: %s", key, ovals[key])

    return ovals


def get_json_data(json_file):
    """
    Retrieves JSON formatted data from a file.
    :param json_file:
    :return: JSON data (dependent on the file loaded, usually a dictionary.)
    """
    logging.debug("Extracting JSON file %s", json_file)
    with open(json_file, "r") as json_d:
        d = json.load(json_d)
    return d


def main(args):
    """
    Main function for parseJSON2Oval.py
    :param args:
    :return:
    """

    if args["verbose"]:
        logging.basicConfig(level=logging.DEBUG)
    else:
        if args["quiet"]:
            logging.basicConfig(level=logging.ERROR)
        else:
            logging.basicConfig(level=logging.WARNING)

    if os.path.isfile(args["tracker_data_file"]):
        dsa_data_file = args["tracker_data_file"]
    else:
        logging.error("Using option -d requires a valid file!")
        sys.exit(1)

    # unpack args
    json_file = args["JSONfile"]
    temp_file = args["tmp"]
    release = args["release"]

    if json_file:
        json_data = get_json_data(json_file)
    else:
        logging.debug("Preparing to download JSONfile")
        if os.path.isfile(temp_file):
            logging.warning("Removing file %s", temp_file)
            os.remove(temp_file)
        logging.debug("Issuing wget for JSON file")
        args = [
            "wget",
            "https://security-tracker.debian.org/tracker/data/json",
            "-O",
            temp_file,
        ]
        if os.path.isdir("/etc/ssl"):
            if os.path.isdir("/etc/ssl/ca-debian"):
                args.insert(1, "--ca-directory=/etc/ssl/ca-debian")
        subprocess.call(args)
        logging.debug("File %s received", temp_file)
        json_data = get_json_data(temp_file)
        if os.path.isfile(temp_file):
            logging.debug("Removing file %s", temp_file)
            os.remove(temp_file)

    ovals = {}
    ovals = parseJSON(ovals, json_data, release)
    logging.info("Finished parsing JSON data")
    #ovals = parsedirs(ovals, data_dir, re.compile("^d[ls]a.+\.data$"), 2, release)
    collect_dsa_data(ovals, release, dsa_data_file)
    printdsas(ovals)


if __name__ == "__main__":
    PARSER = argparse.ArgumentParser(
        description="Generates oval definitions "
        "from the JSON file used to "
        "build the Debian Security "
        "Tracker."
    )
    PARSER.add_argument(
        "-q",
        "--quiet",
        help="Quiet mode",
        action="store_true"
    )
    PARSER.add_argument(
        "-v",
        "--verbose",
        help="Verbose Mode",
        action="store_true"
    )
    PARSER.add_argument(
        "-j",
        "--JSONfile",
        type=str,
        help="Local JSON file to use. This will use a local "
        "copy of the JSON file instead of downloading from"
        " it from the server. default=none",
        default=None,
    )
    PARSER.add_argument(
        "-d",
        "--tracker-data-file",
        type=str,
        help="dsa.data file with Security Tracker information to use. "
        "default=../data/dsa.data",
        default="../data/dsa.data"
    )
    PARSER.add_argument(
        "-t",
        "--tmp",
        type=str,
        help="Temporary file to download JSON file to. Warning:"
        " if this file already exists it will be removed "
        "prior to downloading the JSON file. default= "
        "./DebSecTrackTMP.t",
        default="./DebSecTrackTMP.t",
    )
    PARSER.add_argument(
        "-r",
        "--release",
        type=str,
        help="Limit to this release. default= jessie",
        default="jessie",
    )
    PARSER.add_argument(
        "-i",
        "--id",
        type=int,
        help="id number to start defintions at. default=100",
        default=100,
    )
    ARGS = vars(PARSER.parse_args())
    main(ARGS)