vulnerability_summarizer.py

import re

from addict import Dict

SEVERITIES = ["LOW", "MODERATE", "HIGH", "CRITICAL"]


def get_max_severity(repo):
    max_severity_index = -1
    severity = "NONE"

    for va_edge in repo.vulnerabilityAlerts.edges:

        for v_edge in va_edge.node.securityAdvisory.vulnerabilities.edges:

            edge_severity = v_edge.node.severity
            severity_index = SEVERITIES.index(edge_severity)
            if severity_index > max_severity_index:
                max_severity_index = severity_index
                severity = edge_severity

    return severity


def group_by_severity(repositories):
    vulnerable_by_severity = Dict({sev: [] for sev in SEVERITIES})
    for repo in repositories:
        severity = get_max_severity(repo)
        vulnerable_by_severity[severity].append(repo)
    return vulnerable_by_severity


def get_next_major(major, majors):
    found = False
    next_major = None
    for major2 in majors:
        if not found and major2 > major:
            found = True
            next_major = major2

    return next_major


def get_major_version(version):
    """
    Enable checking that 2 versions are within the same
    major version
    """
    components = re.findall(r"\d+", version)
    major = components[0]
    return major


def get_uniform_version(version):
    """
    Remove any non-numeric characters from required version text
    """
    components = re.findall(r"\d+", version)
    uniform = ".".join(components)
    return uniform


def get_sortable_version(version):
    """
    Add leading zeros to version number components to
    ensure that 1.10.x is > 1.2.x
    """
    version_components = re.findall(r"\d+", version)
    print(version_components)
    sortable = []
    for component in version_components:
        sortable.append(format(int(component), "04d"))

    return ".".join(sortable)


def get_vulnerability_patches(repo):
    patches = []

    # log case when extra pages are found
    if repo.vulnerabilityAlerts.pageInfo.hasNextPage:
        print(f"ERROR: {repo.name} has multiple vulnerabilities")

    for v_edge in repo.vulnerabilityAlerts.edges:
        package = v_edge.node.securityVulnerability.package.name
        print(package)
        dependency_file = v_edge.node.vulnerableManifestPath
        required_version = get_uniform_version(v_edge.node.vulnerableRequirements)
        sortable_version = get_sortable_version(required_version)
        print(f"{required_version} == {sortable_version}")

        patch_sortable_version = None
        patch_version = None
        alternative_patched_versions = []
        patch_found = False
        patchable = False
        max_severity = 0

        # log case when extra pages are found
        if v_edge.node.securityAdvisory.vulnerabilities.pageInfo.hasNextPage:
            print(f"ERROR: {repo.name} {package} has more than 20 advisories")

        advisories = v_edge.node.securityAdvisory.vulnerabilities.edges
        for a_edge in advisories:
            advisory = a_edge.node

            # Some advisories apply to packages that don't match
            # the installed package.
            same_package = advisory.package.name == package

            severity_index = SEVERITIES.index(advisory.severity)
            if severity_index > max_severity:
                max_severity = severity_index

            if advisory.firstPatchedVersion:
                advisory_patch_version = get_uniform_version(
                    advisory.firstPatchedVersion.identifier
                )
                advisory_sortable_version = get_sortable_version(advisory_patch_version)
                print(advisory_sortable_version)
                patch_is_newer = advisory_sortable_version > sortable_version
                patchable = same_package and patch_is_newer
                if patchable:
                    patch_found = True
                    print(f"{advisory_sortable_version} is patchable")
                    alternative_patched_versions.append(
                        Dict(
                            {
                                "version": advisory_patch_version,
                                "sortable": advisory_sortable_version,
                            }
                        )
                    )

                    is_higher_patch = (not patch_version) or (
                        advisory_sortable_version > patch_sortable_version
                    )
                    if is_higher_patch:
                        patch_version = advisory_patch_version
                        patch_sortable_version = advisory_sortable_version

        alternative_patched_versions.sort(key=lambda patch: patch["sortable"])

        if patch_found:
            patches.append(
                Dict(
                    {
                        "package": package,
                        "dependency_file": dependency_file,
                        "current_version": required_version,
                        "patch_available": patch_found,
                        "patch_version": patch_version,
                        "alternative_patched_versions": alternative_patched_versions,
                        "sortable_version": patch_sortable_version,
                        "severity": SEVERITIES[max_severity],
                        "severity_index": max_severity,
                    }
                )
            )
        else:
            patches.append(
                Dict(
                    {
                        "package": package,
                        "dependency_file": dependency_file,
                        "current_version": required_version,
                        "patch_available": patch_found,
                        "patch_version": required_version,
                        "alternative_patched_versions": alternative_patched_versions,
                        "sortable_version": sortable_version,
                        "severity": SEVERITIES[max_severity],
                        "severity_index": max_severity,
                    }
                )
            )

    print(patches)
    return patches


def get_consolidated_alternate_versions(versions1, versions2):
    """
    If multiple vulnerabilities exist for the same package you
    should patch to a version which fixes both.
    This means the higher if the same major version exists in
    both lists.
    """

    # print(versions1)
    # print(versions2)
    alternates = {}
    # get versions indexed by major version
    majors1 = {get_major_version(version.version): version for version in versions1}
    majors2 = {get_major_version(version.version): version for version in versions2}
    # print(majors1)
    # print(majors2)

    # get list of present major versions
    majors1_list = [major for major in majors1.keys()]
    majors2_list = [major for major in majors2.keys()]
    # print(majors1_list)
    # print(majors2_list)
    all_majors = list(set(majors1_list + majors2_list))
    all_majors.sort()
    # print(all_majors)

    for major in all_majors:
        alternate = None
        if major in majors1_list and major in majors2_list:
            if majors1[major]["sortable"] > majors2[major]["sortable"]:
                alternate = majors1[major]
            else:
                alternate = majors2[major]
        elif major in majors1_list and major not in majors2_list:
            next_major = get_next_major(major, majors2_list)
            if next_major:
                alternate = majors2[next_major]

        elif major not in majors1_list and major in majors2_list:
            next_major = get_next_major(major, majors1_list)
            if next_major:
                alternate = majors1[next_major]

        if alternate:
            alternates[alternate.version] = alternate

    alternate_list = [alternate for version, alternate in alternates.items()]
    alternate_list.sort(key=lambda alternate: alternate["sortable"])
    # alternate_versions = [alternate.version for alternate in alternate_list]
    return alternate_list


def get_package_patches(repo):
    """
    Consolidate results if there are multiple vulnerabilities
    for the same package
    """
    vulnerability_patches = get_vulnerability_patches(repo)
    package_patches = Dict({})
    for patch in vulnerability_patches:
        if patch.package in package_patches:
            current_package_patch = package_patches[patch.package]
            current_alternates = current_package_patch.alternative_patched_versions
            new_alternates = patch.alternative_patched_versions

            # Show highest version patch for multiple vulnerabilities
            if patch.sortable_version > current_package_patch.sortable_version:
                package_patches[patch.package] = patch

            # Show highest severity of multiple vulnerabilities
            if current_package_patch.severity_index > patch.severity_index:
                package_patches[patch.package][
                    "severity_index"
                ] = current_package_patch.severity_index
                package_patches[patch.package]["severity"] = SEVERITIES[
                    current_package_patch.severity_index
                ]

            alternatives = get_consolidated_alternate_versions(
                current_alternates, new_alternates
            )

            package_patches[patch.package][
                "alternative_patched_versions"
            ] = alternatives

        else:
            package_patches[patch.package] = patch

        # Remove the recommended version from the alternatives list

        alternatives = []
        for alternate in package_patches[patch.package]["alternative_patched_versions"]:
            if get_uniform_version(patch.patch_version) != get_uniform_version(
                alternate.version
            ):
                alternatives.append(alternate)

        package_patches[patch.package]["alternative_patched_versions"] = alternatives

    print(package_patches)

    return package_patches


def get_patch_list(repo):
    """
    The same package can have multiple alerts or multiple advisories within the same
    alert so to get down to one recommended version per package you have to do an
    aggregation for the advisories in each alert and then to see if there are
    multiple alerts for the same package.

    Also the highest patched version may not correspond to the highest severity
    of all the advisories.
    """
    repo_patches = get_package_patches(repo)
    print(repo_patches)

    # sort by descending severity
    if len(repo_patches.keys()) > 0:
        patch_list = [patch for package, patch in repo_patches.items()]
        patch_list.sort(key=lambda patch: patch["severity_index"], reverse=True)
        print(patch_list)
        patches = {patch["package"]: patch for patch in patch_list}
    else:
        patches = {}

    return patches


def get_repository_severity_counts(repo):
    """
    Count results by severity in repo.patches
    """
    patches = get_patch_list(repo)
    severity_counts = {}
    for package, patch in patches.items():
        severity = patch.severity
        if severity not in severity_counts:
            severity_counts[severity] = 1
        else:
            severity_counts[severity] += 1
    return severity_counts