apache_tomcat_importer.py

#
# Copyright (c) nexB Inc. and others. All rights reserved.
# VulnerableCode is a trademark of nexB Inc.
# SPDX-License-Identifier: Apache-2.0
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
# See https://github.com/aboutcode-org/vulnerablecode for support or download.
# See https://aboutcode.org for more information about nexB OSS projects.
#

import dataclasses
import re
import urllib
from collections import defaultdict
from collections import namedtuple
from typing import Iterable
from typing import List

import requests
from bs4 import BeautifulSoup
from packageurl import PackageURL
from univers.version_constraint import VersionConstraint
from univers.version_range import ApacheVersionRange
from univers.version_range import MavenVersionRange
from univers.versions import MavenVersion
from univers.versions import SemverVersion

from vulnerabilities.importer import AdvisoryDataV2
from vulnerabilities.importer import AffectedPackageV2
from vulnerabilities.importer import ReferenceV2
from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
from vulnerabilities.pipes.advisory import append_patch_classifications

GITHUB_COMMIT_URL_RE = re.compile(
    r"https?://github\.com/apache/tomcat/commit/(?P<commit_hash>[0-9a-f]{5,40})"
)
GITBOX_COMMIT_URL_RE = re.compile(
    r"https?://gitbox\.apache\.org/repos/asf\?p=tomcat\.git;a=commit;h=(?P<commit_hash>[0-9a-f]{5,40})"
)
TOMCAT_VCS_URL = "https://github.com/apache/tomcat"


class ApacheTomcatImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
    """
    Apache HTTPD Importer Pipeline

    This pipeline imports security advisories from the Apache HTTPD project.
    """

    pipeline_id = "apache_tomcat_importer_v2"
    spdx_license_expression = "Apache-2.0"
    license_url = "https://www.apache.org/licenses/LICENSE-2.0"
    base_url = "https://tomcat.apache.org/security"

    precedence = 200

    def fetch_advisory_links(self):
        """
        Yield the URLs of each Tomcat version security-related page.
        Each page link is in the form of `https://tomcat.apache.org/security-10.html`,
        for instance, for v10.
        """
        data = requests.get(self.base_url).content
        soup = BeautifulSoup(data, features="lxml")
        for tag in soup.find_all("a"):
            link = tag.get("href")

            if link and "security-" in link and any(char.isdigit() for char in link):
                yield urllib.parse.urljoin(self.base_url, link)

    @classmethod
    def steps(cls):
        return (cls.collect_and_store_advisories,)

    @classmethod
    def advisories_count(cls):
        return 0

    def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
        for page_url in self.fetch_advisory_links():
            try:
                content = requests.get(page_url).content
                tomcat_advisories = parse_tomcat_security(content)
                self.log(f"Processing {len(tomcat_advisories)} advisories from {page_url}")
                grouped = defaultdict(list)
                for advisory in tomcat_advisories:
                    grouped[advisory.cve].append(advisory)
                for cve, advisory_list in grouped.items():
                    affected_packages = []
                    references = []
                    all_commit_patches = []

                    for advisory in advisory_list:
                        self.log(f"Processing advisory {advisory.cve}")
                        apache_range = to_version_ranges_apache(
                            advisory.affected_versions,
                            advisory.fixed_in,
                            self,
                        )

                        maven_range = to_version_ranges_maven(
                            advisory.affected_versions,
                            advisory.fixed_in,
                        )

                        commit_patches = get_commit_patches(
                            advisory.commit_urls,
                        )
                        all_commit_patches.extend(commit_patches)

                        for ref_url in advisory.reference_urls:
                            references.append(ReferenceV2(url=ref_url))

                        affected_packages.append(
                            AffectedPackageV2(
                                package=PackageURL(type="apache", name="tomcat"),
                                affected_version_range=apache_range,
                                fixed_by_commit_patches=commit_patches,
                            )
                        )

                        affected_packages.append(
                            AffectedPackageV2(
                                package=PackageURL(
                                    type="maven",
                                    namespace="org.apache.tomcat",
                                    name="tomcat",
                                ),
                                affected_version_range=maven_range,
                                fixed_by_commit_patches=commit_patches,
                            )
                        )
                    page_id = page_url.split("/")[-1].replace(".html", "")
                    yield AdvisoryDataV2(
                        advisory_id=f"{page_id}/{cve}",
                        summary=advisory_list[0].summary,
                        affected_packages=affected_packages,
                        references=references,
                        url=page_url,
                        original_advisory_text=str(content),
                    )

            except Exception as e:
                self.log(f"{e!r}")


def to_version_ranges_apache(version_item, fixed_version, self=None):
    constraints = []

    VersionConstraintTuple = namedtuple("VersionConstraintTuple", ["comparator", "version"])
    affected_constraint_tuple_list = []
    fixed_constraint_tuple_list = []

    if version_item:
        version_item = version_item.strip()
        if "to" in version_item:
            version_item_split = version_item.split(" ")
            affected_constraint_tuple_list.append(
                VersionConstraintTuple(">=", version_item_split[0])
            )
            affected_constraint_tuple_list.append(
                VersionConstraintTuple("<=", version_item_split[-1])
            )

        elif "-" in version_item:
            version_item_split = version_item.split("-")
            affected_constraint_tuple_list.append(
                VersionConstraintTuple(">=", version_item_split[0])
            )
            affected_constraint_tuple_list.append(
                VersionConstraintTuple("<=", version_item_split[-1])
            )

        elif version_item.startswith("<"):
            version_item_split = version_item.split("<")
            affected_constraint_tuple_list.append(
                VersionConstraintTuple("<", version_item_split[-1])
            )

        else:
            version_item_split = version_item.split(" ")
            affected_constraint_tuple_list.append(
                VersionConstraintTuple("=", version_item_split[0])
            )

    if fixed_version:
        fixed_item_split = fixed_version.split(" ")
        fixed_constraint_tuple_list.append(VersionConstraintTuple("=", fixed_item_split[0]))

    for record in affected_constraint_tuple_list:
        try:
            constraints.append(
                VersionConstraint(
                    comparator=record.comparator,
                    version=SemverVersion(record.version),
                )
            )
        except Exception as e:
            if self:
                self.log(f"{record.version!r} is not a valid SemverVersion {e!r}")
                continue

    for record in fixed_constraint_tuple_list:
        constraints.append(
            VersionConstraint(
                comparator=record.comparator,
                version=SemverVersion(record.version),
            ).invert()
        )

    return ApacheVersionRange(constraints=constraints)


def to_version_ranges_maven(version_item, fixed_version):
    constraints = []

    if version_item:
        version_item = version_item.strip()
        if "to" in version_item:
            version_item_split = version_item.split(" ")

            constraints.append(
                VersionConstraint(
                    comparator=">=",
                    version=MavenVersion(version_item_split[0]),
                )
            )
            constraints.append(
                VersionConstraint(
                    comparator="<=",
                    version=MavenVersion(version_item_split[-1]),
                )
            )

        elif "-" in version_item:
            version_item_split = version_item.split("-")

            constraints.append(
                VersionConstraint(
                    comparator=">=",
                    version=MavenVersion(version_item_split[0]),
                )
            )
            constraints.append(
                VersionConstraint(
                    comparator="<=",
                    version=MavenVersion(version_item_split[-1]),
                )
            )

        elif version_item.startswith("<"):
            version_item_split = version_item.split("<")

            constraints.append(
                VersionConstraint(
                    comparator="<",
                    version=MavenVersion(version_item_split[-1]),
                )
            )

        else:
            version_item_split = version_item.split(" ")

            constraints.append(
                VersionConstraint(
                    comparator="=",
                    version=MavenVersion(version_item_split[0]),
                )
            )

    if fixed_version:
        fixed_item_split = fixed_version.split(" ")

        constraints.append(
            VersionConstraint(
                comparator="=",
                version=MavenVersion(fixed_item_split[0]),
            ).invert()
        )

    return MavenVersionRange(constraints=constraints)


@dataclasses.dataclass(order=True)
class TomcatAdvisoryData:
    cve: str
    summary: str
    fixed_in: str
    affected_versions: str
    commit_urls: List[str] = dataclasses.field(default_factory=list)
    reference_urls: List[str] = dataclasses.field(default_factory=list)


def parse_tomcat_security(html_content):
    soup = BeautifulSoup(html_content, "lxml")
    results = []

    for header in soup.find_all("h3"):
        header_text = header.get_text(" ", strip=True)
        m = re.search(
            r"(?:Tomcat\s+|Fixed\s+(?:in\s+)?)((?:\d+\.)*\d+)",
            header_text,
            flags=re.IGNORECASE,
        )
        if not m:
            continue
        fixed_in = m.group(1)

        container = header.find_next_sibling("div", class_="text")
        if not container:
            continue

        current = None

        for p in container.find_all("p", recursive=False):

            strong = p.find("strong")
            cve_link = p.find("a", href=re.compile(r"CVE-"))

            if strong and cve_link:
                if current:
                    results.append(_finalize_advisory(current))

                current = {
                    "cve": cve_link.get_text(strip=True),
                    "summary": strong.get_text(" ", strip=True),
                    "affected_versions": None,
                    "fixed_in": fixed_in,
                    "commit_urls": [],
                    "reference_urls": [],
                }
                continue

            if current:
                text = p.get_text(" ", strip=True)

                lower_text = text.lower()
                if "fixed" in lower_text and "commit" in lower_text:
                    for link in p.find_all("a", href=True):
                        href = link["href"]
                        if GITHUB_COMMIT_URL_RE.match(href) or GITBOX_COMMIT_URL_RE.match(href):
                            current["commit_urls"].append(href)
                        current["reference_urls"].append(href)

                if text.startswith("Affects:"):
                    current["affected_versions"] = text.replace("Affects:", "").strip()
                    results.append(_finalize_advisory(current))
                    current = None

        if current:
            results.append(_finalize_advisory(current))

    return results


def _finalize_advisory(current):
    return TomcatAdvisoryData(
        cve=current["cve"],
        summary=current["summary"],
        affected_versions=current["affected_versions"],
        fixed_in=current["fixed_in"],
        commit_urls=current.get("commit_urls", []),
        reference_urls=current.get("reference_urls", []),
    )


def get_commit_patches(commit_urls):
    commit_patches = []
    for url in commit_urls:
        match = GITHUB_COMMIT_URL_RE.match(url) or GITBOX_COMMIT_URL_RE.match(url)
        if not match:
            continue

        commit_hash = match.group("commit_hash")
        classified_packages = []
        append_patch_classifications(
            url=url,
            commit_hash=commit_hash,
            patch_text=None,
            affected_packages=classified_packages,
            patches=[],
            references=[],
        )
        for package in classified_packages:
            for patch in package.fixed_by_commit_patches:
                commit_patches.append(patch)
    return commit_patches