diff --git a/vulnerabilities/improvers/__init__.py b/vulnerabilities/improvers/__init__.py index 9b11c7920..95c1f3345 100644 --- a/vulnerabilities/improvers/__init__.py +++ b/vulnerabilities/improvers/__init__.py @@ -18,6 +18,7 @@ from vulnerabilities.pipelines import enhance_with_kev from vulnerabilities.pipelines import enhance_with_metasploit from vulnerabilities.pipelines import flag_ghost_packages +from vulnerabilities.pipelines.liferay_advisories import LiferayAdvisoryPipeline IMPROVERS_REGISTRY = [ valid_versions.GitHubBasicImprover, @@ -45,6 +46,7 @@ compute_package_version_rank.ComputeVersionRankPipeline, collect_commits.CollectFixCommitsPipeline, add_cvss31_to_CVEs.CVEAdvisoryMappingPipeline, + LiferayAdvisoryPipeline, ] IMPROVERS_REGISTRY = { diff --git a/vulnerabilities/pipelines/liferay_advisories.py b/vulnerabilities/pipelines/liferay_advisories.py new file mode 100644 index 000000000..e4fb3ad73 --- /dev/null +++ b/vulnerabilities/pipelines/liferay_advisories.py @@ -0,0 +1,187 @@ +# 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 re +from typing import Iterable + +import requests +from bs4 import BeautifulSoup +from packageurl import PackageURL +from requests.exceptions import HTTPError +from requests.exceptions import RequestException +from requests.exceptions import Timeout + +from vulnerabilities.importer import AdvisoryData +from vulnerabilities.importer import AffectedPackage +from vulnerabilities.importer import Reference +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline + + +class LiferayAdvisoryPipeline(VulnerableCodeBaseImporterPipeline): + """Pipeline to import Liferay security advisories.""" + + pipeline_id = "liferay_advisories" + description = "Import Liferay security advisories" + license_url = "https://liferay.dev/portal/security/known-vulnerabilities" + spdx_license_expression = "CC-BY-4.0" + importer_name = "Liferay Security Advisories" + + @classmethod + def steps(cls): + return ( + cls.fetch_advisories, + cls.parse_advisories, + cls.collect_and_store_advisories, # Changed from collect_advisories + cls.import_new_advisories, + ) + + def fetch_advisories(self): + """Fetch HTML content from Liferay's security page.""" + try: + response = requests.get( + self.license_url, + headers={"User-Agent": "Mozilla/5.0"}, + timeout=30, + ) + response.raise_for_status() + self.html_content = response.text + self.log(f"Response size: {len(self.html_content)} bytes") + except (Timeout, HTTPError, RequestException) as e: + self.log(f"Request error: {e}") + raise + except Exception as e: + self.log(f"Unexpected error: {e}") + raise + + def parse_advisories(self): + """Parse HTML to extract advisories.""" + self.parsed_advisories = [] + soup = BeautifulSoup(self.html_content, "html.parser") + + # Look for CVE IDs anywhere in the page + cve_patterns = soup.find_all(text=re.compile(r"CVE-\d{4}-\d+", re.IGNORECASE)) + self.log(f"Found {len(cve_patterns)} potential CVE mentions") + + if not cve_patterns: + self.log("No CVE IDs found in the page") + return + + # Process each CVE mention + processed_cves = set() + for cve_text in cve_patterns: + # Extract the CVE ID + match = re.search(r"(CVE-\d{4}-\d+)", cve_text, re.IGNORECASE) + if not match: + continue + + cve_id = match.group(1).upper() + if cve_id in processed_cves: + continue + + processed_cves.add(cve_id) + + # Find the parent container + parent = cve_text.parent + container = None + for _ in range(5): + if not parent: + break + if parent.name in ["article", "section", "div"] and len(parent.get_text()) > 100: + container = parent + break + parent = parent.parent + + if not container: + continue + + # Extract information + try: + # Description - get all text from paragraphs + paragraphs = container.find_all("p") + description = " ".join(p.get_text(strip=True) for p in paragraphs) + + # Affected versions + affected_versions = [] + version_section = container.find( + text=re.compile(r"affected|versions", re.IGNORECASE) + ) + + if version_section: + version_list = None + parent = version_section.parent + for _ in range(3): + if not parent: + break + lists = parent.find_all(["ul", "ol"]) + if lists: + version_list = lists[0] + break + parent = parent.parent + + if version_list: + affected_versions = [ + li.get_text(strip=True) for li in version_list.find_all("li") + ] + + # References - all links in the container + references = [] + for a in container.find_all("a", href=True): + if a["href"].startswith("http"): + references.append(a["href"]) + + self.parsed_advisories.append( + { + "cve_id": cve_id, + "summary": description[:500] if description else "", + "affected_versions": affected_versions, + "references": references, + } + ) + + self.log(f"Successfully parsed advisory for {cve_id}") + + except Exception as e: + self.log(f"Error parsing advisory for {cve_id}: {e}") + + def parse_version_to_purl(self, version_text): + """Convert version text to a PackageURL.""" + version_text = version_text.strip() + if "DXP" in version_text.upper(): + match = re.search(r"\d+(?:\.\d+)*", version_text) + if not match: + raise ValueError(f"No version found in {version_text}") + return PackageURL(type="liferay", name="dxp", version=match.group()) + else: + match = re.search(r"\d+(?:\.\d+)*", version_text) + if not match: + raise ValueError(f"No version found in {version_text}") + return PackageURL(type="liferay", name="portal", version=match.group()) + + def collect_advisories(self) -> Iterable[AdvisoryData]: + """Generate advisory data from parsed content.""" + for advisory in self.parsed_advisories: + affected_packages = [] + for version in advisory.get("affected_versions", []): + try: + purl = self.parse_version_to_purl(version) + affected_packages.append(AffectedPackage(package=purl)) + except ValueError as e: + self.log(f"Skipping invalid version {version}: {e}") + + yield AdvisoryData( + aliases=[advisory["cve_id"]], + summary=advisory["summary"], + references=[Reference(url=url) for url in advisory.get("references", [])], + affected_packages=affected_packages, + url=self.license_url, + ) + + def advisories_count(self): + if hasattr(self, "parsed_advisories"): + return len(self.parsed_advisories) + return 0