__init__.py

# Copyright (C) 2021 Intel Corporation
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import annotations

import csv
import json
import os
import time
from collections import defaultdict
from datetime import datetime
from logging import Logger
from pathlib import Path

from cve_bin_tool.cve_scanner import CVEData
from cve_bin_tool.cvedb import CVEDB, DISK_LOCATION_DEFAULT
from cve_bin_tool.error_handler import ErrorHandler, ErrorMode
from cve_bin_tool.log import LOGGER
from cve_bin_tool.output_engine.console import output_console
from cve_bin_tool.output_engine.html import output_html
from cve_bin_tool.output_engine.json_output import output_json, output_json2
from cve_bin_tool.output_engine.util import (
    ProductInfo,
    Remarks,
    VersionInfo,
    add_extension_if_not,
    format_output,
    format_path,
    format_version_range,
    generate_filename,
    get_cve_summary,
    intermediate_output,
)
from cve_bin_tool.sbom_manager.generate import SBOMGenerate
from cve_bin_tool.version import VERSION
from cve_bin_tool.vex_manager.generate import VEXGenerate


def save_intermediate(
    all_cve_data: dict[ProductInfo, CVEData],
    filename: str,
    tag: str,
    scanned_dir: str,
    products_with_cve: int,
    products_without_cve: int,
    total_files: int,
):
    """Save the intermediate report"""

    inter_output = intermediate_output(
        all_cve_data,
        tag,
        scanned_dir,
        products_with_cve,
        products_without_cve,
        total_files,
    )
    with open(filename, "w") as f:
        json.dump(inter_output, f, indent="    ")


def output_csv(
    all_cve_data: dict[ProductInfo, CVEData],
    all_cve_version_info: dict[str, VersionInfo] | None,
    outfile,
    detailed: bool = False,
    affected_versions: int = 0,
    metrics: bool = False,
):
    """Output a CSV of CVEs"""
    formatted_output = format_output(
        all_cve_data,
        all_cve_version_info,
        detailed,
        affected_versions,
        metrics,
    )

    # Remove triage response and justification from the CSV output.
    for cve_entry in formatted_output:
        cve_entry.pop("response", None)
        cve_entry.pop("justification", None)

    # Trim any leading -, =, +, @, tab or CR to avoid excel macros
    for cve_entry in formatted_output:
        for key, value in cve_entry.items():
            cve_entry[key] = value.strip("-=+@\t\r")

    fieldnames = [
        "vendor",
        "product",
        "version",
        "location",
        "cve_number",
        "severity",
        "score",
        "source",
        "cvss_version",
        "cvss_vector",
        "paths",
        "remarks",
        "comments",
    ]
    if metrics:
        fieldnames.append("epss_probability")
        fieldnames.append("epss_percentile")
    if detailed:
        fieldnames.append("description")
    if affected_versions != 0:
        fieldnames.append("affected_versions")
    writer = csv.DictWriter(outfile, fieldnames=fieldnames)

    writer.writeheader()
    writer.writerows(formatted_output)


# load pdfs only if reportlab is found.  if not, make a stub that prints a
# logger message
try:
    from . import pdfbuilder

    def output_pdf(
        all_cve_data: dict[ProductInfo, CVEData],
        is_report,
        products_with_cve,
        all_cve_version_info,
        outfile,
        merge_report,
        affected_versions: int = 0,
        exploits: bool = False,
        metrics: bool = False,
        all_product_data=None,
        cachedir: str = DISK_LOCATION_DEFAULT,
    ):
        """Output a PDF of CVEs"""
        cvedb_data = CVEDB(cachedir=cachedir)
        db_date = time.strftime(
            "%d %B %Y at %H:%M:%S", time.localtime(cvedb_data.get_db_update_date())
        )
        app_version = VERSION
        # Build document
        pdfdoc = pdfbuilder.PDFBuilder()
        cm = pdfdoc.cm
        severity_colour = {
            "UNKNOWN": pdfdoc.grey,
            "LOW": pdfdoc.green,
            "MEDIUM": pdfdoc.orange,
            "HIGH": pdfdoc.blue,
            "CRITICAL": pdfdoc.red,
        }
        pdfdoc.front_page("Vulnerability Report")
        pdfdoc.heading(1, "Introduction")
        pdfdoc.paragraph(
            "The identification of vulnerabilities has been performed using cve-bin-tool version "
            + app_version
        )
        pdfdoc.paragraph(
            "The data used in the creation of this report uses data from a number of data sources including the National Vulnerability Database (NVD),"
            " the Open Source Vulnerability Database (OSV) and Gitlab Advisory Database (GAD). The cve-bin-tool is not endorsed or certified by the NVD or"
            " other providers of data used by the tool."
        )
        if merge_report:
            pdfdoc.paragraph(
                "The report has been generated by merging multiple intermediate reports."
            )
        else:
            pdfdoc.paragraph(
                "The data used has been obtained from the NVD database which was retrieved on "
                + db_date
                + " and contained "
                + str(cvedb_data.get_cve_count())
                + " entries."
            )

        if is_report:
            pdfdoc.heading(1, "List of All Scanned binaries")
            pdfdoc.createtable(
                "Productlist",
                ["Vendor", "Product", "Version"],
                pdfdoc.tblStyle,
            )
            row = 1
            star_warn = False
            for product_info, cve_data in all_cve_data.items():
                star_warn = True if "*" in product_info.vendor else False
                for cve in cve_data["cves"]:
                    entry = [
                        product_info.vendor,
                        product_info.product,
                        product_info.version,
                    ]
                    pdfdoc.addrow(
                        "Productlist",
                        entry,
                    )
                    row += 1
            pdfdoc.showtable("Productlist", widths=[3 * cm, 2 * cm, 2 * cm])
            pdfdoc.paragraph("* vendors guessed by the tool") if star_warn else None
            pdfdoc.paragraph(
                f"There are {products_with_cve} products with vulnerabilities found."
            )

            pdfdoc.pagebreak()

        if merge_report:
            pdfdoc.heading(1, "Intermediate Reports")
            pdfdoc.paragraph(
                "The following table contains severity levels count of individual intermediate report sorted on the basis of timestamp."
            )
            pdfdoc.createtable(
                "SeverityLevels",
                [
                    "Timestamp",
                    "Tag",
                    "Total\nFiles",
                    "Products\nwith CVE",
                    "Products\nwithout CVE",
                    "UNKNOWN",
                    "LOW",
                    "MEDIUM",
                    "HIGH",
                    "CRITICAL",
                ],
                pdfdoc.intermediateStyle,
            )

            for inter_file in merge_report.intermediate_cve_data:
                entry = [
                    datetime.strptime(
                        inter_file["metadata"]["timestamp"], "%Y-%m-%d.%H-%M-%S"
                    ).strftime("%Y-%m-%d %H:%M"),
                    inter_file["metadata"]["tag"],
                    inter_file["metadata"]["total_files"],
                    inter_file["metadata"]["products_with_cve"],
                    inter_file["metadata"]["products_without_cve"],
                    inter_file["metadata"]["severity"]["UNKNOWN"],
                    inter_file["metadata"]["severity"]["LOW"],
                    inter_file["metadata"]["severity"]["MEDIUM"],
                    inter_file["metadata"]["severity"]["HIGH"],
                    inter_file["metadata"]["severity"]["CRITICAL"],
                ]
                pdfdoc.addrow(
                    "SeverityLevels",
                    entry,
                )
            pdfdoc.showtable(
                "SeverityLevels",
                widths=[
                    2.5 * cm,
                    3 * cm,
                    1.5 * cm,
                    2.5 * cm,
                    2.5 * cm,
                    None,
                    None,
                    None,
                    None,
                    None,
                ],
            )
            pdfdoc.pagebreak()

        if products_with_cve != 0:
            pdfdoc.heading(1, "Summary of Identified Vulnerabilities")

            pdfdoc.paragraph("A summary of the vulnerabilities found.")
            pdfdoc.createtable("CVESummary", ["Severity", "Count"], pdfdoc.tblStyle)
            summary = get_cve_summary(all_cve_data, exploits)
            row = 1
            for severity, count in summary.items():
                pdfdoc.addrow(
                    "CVESummary",
                    [severity, count],
                    [
                        (
                            "TEXTCOLOR",
                            (0, row),
                            (1, row),
                            severity_colour[severity.split("-")[0].upper()],
                        ),
                        ("FONT", (0, row), (1, row), "Helvetica-Bold"),
                    ],
                )
                row += 1
            pdfdoc.showtable("CVESummary", widths=[5 * cm, 2 * cm])

            pdfdoc.heading(1, "List of Identified Vulnerabilities")
            pdfdoc.paragraph(
                "The following vulnerabilities are reported against the identified versions of the components."
            )
            col_headings = [
                "Vendor",
                "Product",
                "Version",
                "CVE Number",
                "Source",
                "Severity",
            ]
            col_dist = [10, 10, None, None, None, None]
            if affected_versions != 0:
                col_headings.append("Affected Versions")
                col_dist.append(None)

            cve_by_remarks: defaultdict[Remarks, list[dict[str, str]]] = defaultdict(
                list
            )
            cve_by_paths: defaultdict[Remarks, list[dict[str, str]]] = defaultdict(list)
            # group cve_data by its remarks and separately by paths
            for product_info, cve_data in all_cve_data.items():
                for cve in cve_data["cves"]:
                    if isinstance(cve, str):
                        continue
                    cve_by_remarks[cve.remarks].append(
                        {
                            "vendor": product_info.vendor,
                            "product": product_info.product,
                            "version": product_info.version,
                            "cve_number": cve.cve_number,
                            "source": cve.data_source,
                            "severity": cve.severity,
                            "score": str(cve.score),
                            "cvss_version": str(cve.cvss_version),
                            "comments": cve.comments,
                        }
                    )
                    path_elements = ", ".join(cve_data["paths"])
                    for path_element in path_elements.split(","):
                        path_entry = {
                            "vendor": product_info.vendor,
                            "product": product_info.product,
                            "version": product_info.version,
                            "paths": path_element,
                        }
                        if path_entry not in cve_by_paths[cve.remarks]:
                            cve_by_paths[cve.remarks].append(path_entry)
                    if affected_versions != 0:
                        try:
                            version_info = all_cve_version_info[cve.cve_number]
                        except (
                            KeyError
                        ):  # TODO: handle 'UNKNOWN' and some cves more cleanly
                            version_info = VersionInfo("", "", "", "")
                        cve_by_remarks[cve.remarks][-1].update(
                            {"affected_versions": format_version_range(version_info)}
                        )

            for remarks in sorted(cve_by_remarks):
                row = 1
                star_warn = False
                pdfdoc.createtable(
                    "Productlist",
                    col_headings,
                    pdfdoc.tblStyle,
                    col_dist,
                )
                for cve_data_by_remarks in cve_by_remarks[remarks]:
                    if cve_data_by_remarks["cve_number"] != "UNKNOWN":
                        if "*" in cve_data_by_remarks["vendor"]:
                            star_warn = True
                        entry = [
                            cve_data_by_remarks["vendor"],
                            cve_data_by_remarks["product"],
                            cve_data_by_remarks["version"],
                            cve_data_by_remarks["cve_number"],
                            cve_data_by_remarks["source"],
                            cve_data_by_remarks["severity"],
                        ]
                        if affected_versions != 0:
                            entry.append(cve_data_by_remarks["affected_versions"])
                        pdfdoc.addrow(
                            "Productlist",
                            entry,
                            [
                                (
                                    "TEXTCOLOR",
                                    (0, row),
                                    (2, row),
                                    pdfdoc.black,
                                ),
                                ("FONT", (0, row), (2, row), "Helvetica"),
                                (
                                    "TEXTCOLOR",
                                    (3, row),
                                    (5, row),
                                    severity_colour[
                                        cve_data_by_remarks["severity"]
                                        .split("-")[0]
                                        .upper()
                                    ],
                                ),
                                ("FONT", (3, row), (5, row), "Helvetica-Bold"),
                            ],
                        )
                        row += 1
                        if cve_data_by_remarks["comments"] != "":
                            entry[0] = "Comment:"
                            entry[1] = ""
                            entry[2] = ""
                            entry[3] = cve_data_by_remarks["comments"]
                            entry[4] = ""
                            pdfdoc.addrow(
                                "Productlist",
                                entry,
                                [
                                    (
                                        "TEXTCOLOR",
                                        (0, row),
                                        (5, row),
                                        pdfdoc.black,
                                    ),
                                    ("FONT", (0, row), (2, row), "Helvetica"),
                                    ("FONT", (3, row), (5, row), "Helvetica-Bold"),
                                ],
                            )
                            row += 1

                widths = [3 * cm, 3 * cm, 2 * cm, 4 * cm, 2 * cm, 3 * cm]
                if affected_versions != 0:
                    widths.append(4 * cm)
                pdfdoc.heading(2, remarks.name + " CVEs")
                pdfdoc.showtable("Productlist", widths=widths)
                pdfdoc.paragraph("* vendors guessed by the tool") if star_warn else None

                pdfdoc.paragraph(
                    "The following vulnerabilities are reported against the identified components which contain the reported components."
                )
                pdfdoc.createtable(
                    "Applicationlist",
                    ["Vendor", "Product", "Version", "Root", "Filename"],
                    pdfdoc.tblStyle,
                    [10, 10, 10, 15, 15],
                )
                row = 1
                application_elements = []
                for cve_data_by_paths in cve_by_paths[remarks]:
                    path_root = format_path(cve_data_by_paths["paths"])
                    cve_path_entry = [
                        cve_data_by_paths["vendor"],
                        cve_data_by_paths["product"],
                        cve_data_by_paths["version"],
                        path_root[0],
                        path_root[1],
                    ]
                    if cve_path_entry not in application_elements:
                        application_elements.append(cve_path_entry)
                        pdfdoc.addrow(
                            "Applicationlist",
                            cve_path_entry,
                            [
                                (
                                    "TEXTCOLOR",
                                    (0, row),
                                    (2, row),
                                    pdfdoc.black,
                                ),
                                ("FONT", (0, row), (2, row), "Helvetica"),
                                (
                                    "TEXTCOLOR",
                                    (3, row),
                                    (4, row),
                                    pdfdoc.black,
                                ),
                                ("FONT", (3, row), (5, row), "Helvetica-Bold"),
                            ],
                        )
                        row += 1
                pdfdoc.showtable(
                    "Applicationlist", widths=[3 * cm, 3 * cm, 2 * cm, 4 * cm, 3 * cm]
                )

            if metrics:
                pdfdoc.heading(1, "List of Vulnerabilities with different metric")
                pdfdoc.paragraph(
                    "The table given below gives CVE found with there score on different metrics."
                )
                cve_by_metrics: defaultdict[Remarks, list[dict[str, str]]] = (
                    defaultdict(list)
                )
                col_headings = [
                    "CVE Number",
                    "CVSS_version",
                    "CVSS_score",
                    "EPSS_probability",
                    "EPSS_percentile",
                ]
                # group cve_data by its remarks and separately by paths
                for product_info, cve_data in all_cve_data.items():
                    for cve in cve_data["cves"]:
                        probability = "-"
                        percentile = "-"
                        for metric, field in cve.metric.items():
                            if metric == "EPSS":
                                probability = round(field[0], 5)
                                percentile = field[1]

                        cve_by_metrics[cve.remarks].append(
                            {
                                "cve_number": cve.cve_number,
                                "cvss_version": str(cve.cvss_version),
                                "cvss_score": str(cve.score),
                                "epss_probability": str(probability),
                                "epss_percentile": str(percentile),
                                "severity": cve.severity,
                            }
                        )

                for remarks in sorted(cve_by_metrics):
                    pdfdoc.createtable(
                        "cvemetric",
                        col_headings,
                        pdfdoc.tblStyle,
                    )
                    row = 1
                    for cve in cve_by_metrics[remarks]:
                        entry = [
                            cve["cve_number"],
                            cve["cvss_version"],
                            str(cve["cvss_score"]),
                            str(cve["epss_probability"]),
                            str(cve["epss_percentile"]),
                        ]
                        pdfdoc.addrow(
                            "cvemetric",
                            entry,
                            [
                                (
                                    "TEXTCOLOR",
                                    (0, row),
                                    (4, row),
                                    severity_colour[
                                        cve["severity"].split("-")[0].upper()
                                    ],
                                ),
                                ("FONT", (0, row), (4, row), "Helvetica-Bold"),
                            ],
                        )
                        row += 1
                pdfdoc.showtable(
                    "cvemetric", widths=[4 * cm, 4 * cm, 3 * cm, 4 * cm, 4 * cm]
                )

            # List of scanned products with no identified vulnerabilities
            if all_product_data is not None:
                pdfdoc.heading(1, "No Identified Vulnerabilities")
                pdfdoc.paragraph(
                    "The following products were identified as having no identified vulnerabilities."
                )
                pdfdoc.createtable(
                    "Productlist",
                    ["Vendor", "Product", "Version"],
                    pdfdoc.tblStyle,
                    [10, 10, 10],
                )
                row = 1
                products_with_cves = list(map(lambda x: x[1], all_cve_data))
                for product_data in all_product_data:
                    if (
                        all_product_data[product_data] == 0
                        and product_data.product not in products_with_cves
                    ):
                        product_entry = [
                            product_data.vendor,
                            product_data.product,
                            product_data.version,
                        ]
                        pdfdoc.addrow(
                            "Productlist",
                            product_entry,
                            [
                                (
                                    "TEXTCOLOR",
                                    (0, row),
                                    (2, row),
                                    pdfdoc.black,
                                ),
                                ("FONT", (0, row), (2, row), "Helvetica"),
                            ],
                        )
                        row += 1

                pdfdoc.showtable("Productlist", widths=[3 * cm, 3 * cm, 2 * cm])

        pdfdoc.pagebreak()
        pdfdoc.paragraph("END OF DOCUMENT.")
        pdfdoc.publish(outfile)

except ModuleNotFoundError:

    def output_pdf(
        all_cve_data: dict[ProductInfo, CVEData],
        is_report,
        products_with_cve,
        all_cve_version_info,
        outfile,
        merge_report,
        affected_versions: int = 0,
        exploits: bool = False,
        all_product_data=None,
        cachedir: str = DISK_LOCATION_DEFAULT
    ):
        """Output a PDF of CVEs
        Required module: Reportlab not found"""
        LOGGER.warning("PDF output requires install of reportlab")


class OutputEngine:
    """
    Class represention of OutputEngine
    Attributes:
        all_cve_data (dict[ProductInfo, CVEData])
        scanned_dir (str)
        filename (str)
        themes_dir (str)
        time_of_last_update
        tag (str)
        logger
        products_with_cve (int)
        products_without_cve (int)
        total_files (int)
        is_report (bool)
        no_zero_cve_report (bool)
        append
        merge_report
        affected_versions (int)
        all_cve_version_info
        detailed (bool)
        vex_filename (str)
        exploits (bool)
        metrics (bool)
        all_product_data
        sbom_filename (str)
        sbom_type (str)
        sbom_format (str)
        sbom_root (str)
        offline (bool)

    Methods:
        output_cves: Output a list of CVEs to other formats like CSV or JSON.
        generate_vex: Generate a vex file and create vulnerability entry.
        generate_sbom: Create SBOM package and generate SBOM file.
        output_file_wrapper:
        output_file: Generate a file for list of CVE
        check_file_path: Generate a new filename if file already exists.
        check_dir_path: Generate a new filename if filepath is a directory.
    """

    def __init__(
        self,
        all_cve_data: dict[ProductInfo, CVEData],
        scanned_dir: str,
        filename: str,
        themes_dir: str,
        time_of_last_update,
        tag: str,
        logger: Logger | None = None,
        products_with_cve: int = 0,
        products_without_cve: int = 0,
        total_files: int = 0,
        is_report: bool = False,
        no_zero_cve_report: bool = False,
        append: str | bool = False,
        merge_report: None | list[str] = None,
        affected_versions: int = 0,
        all_cve_version_info=None,
        detailed: bool = False,
        exploits: bool = False,
        metrics: bool = False,
        all_product_data=None,
        sbom_filename: str = "",
        sbom_type: str = "spdx",
        sbom_format: str = "tag",
        sbom_root: str = "CVE_SBOM",
        vex_filename: str = "",
        vex_type: str = "",
        vex_product_info: dict[str, str] = {},
        offline: bool = False,
        organized_arguements: dict = None,
        cachedir: str = DISK_LOCATION_DEFAULT
    ):
        """Constructor for OutputEngine class."""
        self.logger = logger or LOGGER.getChild(self.__class__.__name__)
        self.all_cve_version_info = all_cve_version_info
        self.scanned_dir = scanned_dir
        self.filename = str(Path(filename).resolve()) if filename else ""
        self.products_with_cve = products_with_cve
        self.products_without_cve = products_without_cve
        self.total_files = total_files
        self.themes_dir = themes_dir
        self.is_report = is_report
        self.no_zero_cve_report = no_zero_cve_report
        self.time_of_last_update = time_of_last_update
        self.append = append
        self.tag = tag
        self.merge_report = merge_report
        self.affected_versions = affected_versions
        self.all_cve_data = all_cve_data
        self.detailed = detailed
        self.exploits = exploits
        self.metrics = metrics
        self.all_product_data = all_product_data
        self.sbom_filename = sbom_filename
        self.sbom_type = sbom_type
        self.sbom_format = sbom_format
        self.sbom_root = sbom_root
        self.offline = offline
        self.organized_arguements = organized_arguements
        self.sbom_packages = {}
        self.vex_type = vex_type
        self.vex_product_info = vex_product_info
        self.vex_filename = vex_filename
        self.cachedir = cachedir

    def output_cves(self, outfile, output_type="console"):
        """Output a list of CVEs
        format self.checkers[checker_name][version] = dict{id: severity}
        to other formats like CSV or JSON
        """
        if output_type == "json":
            output_json(
                self.all_cve_data,
                self.all_cve_version_info,
                outfile,
                self.detailed,
                self.affected_versions,
                self.metrics,
            )
        elif output_type == "json2":
            output_json2(
                self.all_cve_data,
                self.all_cve_version_info,
                self.time_of_last_update,
                outfile,
                self.affected_versions,
                self.organized_arguements,
                self.detailed,
                self.exploits,
                self.metrics,
            )
        elif output_type == "csv":
            output_csv(
                self.all_cve_data,
                self.all_cve_version_info,
                outfile,
                self.detailed,
                self.affected_versions,
                self.metrics,
            )
        elif output_type == "pdf":
            output_pdf(
                self.all_cve_data,
                self.is_report,
                self.products_with_cve,
                self.all_cve_version_info,
                outfile,
                self.merge_report,
                self.affected_versions,
                self.exploits,
                self.metrics,
                self.cachedir,
            )
        elif output_type == "html":
            output_html(
                self.all_cve_data,
                self.all_cve_version_info,
                self.scanned_dir,
                self.filename,
                self.themes_dir,
                self.total_files,
                self.products_with_cve,
                self.products_without_cve,
                self.merge_report,
                self.logger,
                outfile,
                self.affected_versions,
            )
        else:  # console, or anything else that is unrecognised
            output_console(
                self.all_cve_data,
                self.all_cve_version_info,
                self.time_of_last_update,
                self.affected_versions,
                self.exploits,
                self.metrics,
                self.all_product_data,
                self.offline,
                None,
                outfile,
            )

        if isinstance(self.append, str):
            save_intermediate(
                self.all_cve_data,
                self.append,
                self.tag,
                self.scanned_dir,
                self.products_with_cve,
                self.products_without_cve,
                self.total_files,
            )
            self.logger.info(f"Output stored at {self.append}")

        if self.vex_filename != "" or self.vex_type != "":
            vexgen = VEXGenerate(
                self.vex_product_info["product"],
                self.vex_product_info["release"],
                self.vex_product_info["vendor"],
                self.vex_filename,
                self.vex_type,
                self.all_cve_data,
                self.vex_product_info["revision_reason"],
                self.vex_product_info.get("sbom_serial_number", ""),
                logger=self.logger,
            )
            vexgen.generate_vex()
        if self.sbom_filename != "":
            # cyclonedx doesn't support tag or yaml and lib4sbom won't read back the file if it doesn't end with json
            if self.sbom_type == "cyclonedx":
                self.sbom_filename = add_extension_if_not(self.sbom_filename, "json")
            sbomgen = SBOMGenerate(
                self.all_product_data,
                self.sbom_filename,
                self.sbom_type,
                self.sbom_format,
                self.sbom_root,
                self.logger,
            )
            sbomgen.generate_sbom()

    def output_file_wrapper(self, output_types=["console"]):
        """Call output_file method for all output types."""
        for output_type in output_types:
            self.output_file(output_type)

    def output_file(self, output_type="console"):
        """Generate a file for list of CVE"""

        if self.append:
            if isinstance(self.append, str):
                self.append = self.check_dir_path(
                    self.append, output_type="json", prefix="intermediate"
                )
                self.append = add_extension_if_not(self.append, "json")
                self.append = self.check_file_path(
                    self.append, output_type="json", prefix="intermediate"
                )
            else:
                # file path for intermediate report not given
                self.append = generate_filename("json", "intermediate")

        if output_type == "console":
            # short circuit file opening logic if we are actually
            # just writing to stdout
            if self.filename:
                self.filename = add_extension_if_not(self.filename, "txt")
                self.logger.info(f"Console output stored at {self.filename}")
            self.output_cves(self.filename, output_type)
            return

        # Abort processing if no data to write to file and no_zero_cve_report is True
        if len(self.all_cve_data) == 0 and self.no_zero_cve_report:
            return
        # Check if we need to generate a filename
        if not self.filename:
            self.filename = generate_filename(output_type)
        else:
            # check and add if the filename doesn't contain extension
            self.filename = add_extension_if_not(self.filename, output_type)

            self.filename = self.check_file_path(self.filename, output_type)

            # try opening that file
            with ErrorHandler(mode=ErrorMode.Ignore) as e:
                with open(self.filename, "w") as f:
                    f.write("testing")
                os.remove(self.filename)
            if e.exit_code:
                self.logger.info(
                    f"Exception {e.exc_val} occurred while writing to the file {self.filename} "
                    "Switching Back to Default Naming Convention"
                )
                self.filename = generate_filename(output_type)
        # if extension is set to .json2 due to current code logic make it .json
        if self.filename.endswith(".json2"):
            self.filename = self.filename[:-1]
        # Log the filename generated
        self.logger.info(f"{output_type.upper()} report stored at {self.filename}")

        # call to output_cves
        if output_type == "pdf":
            with open(self.filename, "wb") as f:
                self.output_cves(f, output_type)
        else:
            # if type is csv, file should be opened with newline=''
            # see https://docs.python.org/3/library/csv.html#csv.writer
            newline = "" if output_type == "csv" else None
            with open(self.filename, mode="w", newline=newline, encoding="utf8") as f:
                self.output_cves(f, output_type)

    def check_file_path(self, filepath: str, output_type: str, prefix: str = "output"):
        """Generate a new filename if file already exists."""
        # check if the file already exists
        if filepath.endswith(".json2"):
            filepath = filepath[:-1]
        if Path(filepath).is_file():
            self.logger.warning(f"Failed to write at '{filepath}'. File already exists")
            self.logger.info("Generating a new filename with Default Naming Convention")
            filepath = generate_filename(output_type, prefix)

        return filepath

    def check_dir_path(
        self, filepath: str, output_type: str, prefix: str = "intermediate"
    ):
        """Generate a new filename if filepath is a directory."""
        if Path(filepath).is_dir():
            self.logger.info(
                f"Generating a new filename with Default Naming Convention in directory path {filepath}"
            )
            filepath = str(Path(filepath) / generate_filename(output_type, prefix))

        return filepath