From fc05de79e735b26650e7470df9cc84cc04a662bb Mon Sep 17 00:00:00 2001 From: Shrish0098 Date: Tue, 4 Mar 2025 21:26:25 +0530 Subject: [PATCH 1/3] Added Liferay advisories Signed-off-by: Shrish Mishra shrish409@gmail.com Signed-off-by: Shrish0098 --- vulnerabilities/improvers/__init__.py | 2 + .../pipelines/liferay_advisories.py | 157 ++++++++++++++++++ 2 files changed, 159 insertions(+) create mode 100644 vulnerabilities/pipelines/liferay_advisories.py 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..2c5a19f82 --- /dev/null +++ b/vulnerabilities/pipelines/liferay_advisories.py @@ -0,0 +1,157 @@ +import requests +from bs4 import BeautifulSoup +from django.db import transaction +from fake_useragent import UserAgent +from packageurl import PackageURL +from requests.exceptions import HTTPError +from requests.exceptions import RequestException +from requests.exceptions import Timeout + +from vulnerabilities.models import Advisory +from vulnerabilities.models import Vulnerability +from vulnerabilities.models import VulnerabilityReference +from vulnerabilities.pipelines import VulnerableCodePipeline + + +class LiferayAdvisoryPipeline(VulnerableCodePipeline): + 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" + + @classmethod + def steps(cls): + return ( + cls.fetch_advisories, + cls.parse_advisories, + cls.import_to_db, + ) + + def fetch_advisories(self): + try: + ua = UserAgent() + response = requests.get( + "https://liferay.dev/portal/security/known-vulnerabilities", + headers={"User-Agent": ua.chrome}, + timeout=30, + ) + response.raise_for_status() + + if not response.text.strip(): + raise ValueError("Empty response from server") + + self.html_content = response.text + + except Timeout as te: + self.log(f"Timeout occurred: {te}") + raise + except HTTPError as he: + self.log(f"HTTP Error {he.response.status_code}: {he}") + if he.response.status_code == 403: + self.log("Consider rotating User-Agent headers") + raise + except RequestException as re: + self.log(f"Request failed: {re}") + raise + except Exception as e: + self.log(f"Unexpected error: {e}") + raise RuntimeError(f"Fatal pipeline error: {e}") from e + + def parse_advisories(self): + self.parsed_advisories = [] + soup = BeautifulSoup(self.html_content, "html.parser") + + portlet_section = soup.select_one("section.portlet") + if not portlet_section: + self.log("No portlet section found") + return + + # Loop over advisories + for asset_entry in portlet_section.select(".asset-entry, .entry, .asset-publisher"): + try: + cve_id = asset_entry.select_one("h1, h2, h3, .entry-title").get_text(strip=True) + except AttributeError: + self.log("No title found; skipping entry.") + continue + + if not cve_id.startswith("CVE-"): + self.log(f"Skipping non-CVE entry: {cve_id}") + continue + + try: + description = asset_entry.select_one(".entry-content, .asset-content").get_text( + " ", strip=True + ) + + metadata = asset_entry.select_one(".metadata, .entry-metadata") + if not metadata: + self.log(f"No metadata found for {cve_id}; skipping.") + continue + + # Attempt to parse severity and versions + severity_tag = metadata.select_one(".severity:contains('Severity') + dd") + severity = severity_tag.get_text(strip=True) if severity_tag else "Unknown" + + affected_versions = [ + li.get_text(strip=True) + for li in metadata.select(".affected-versions li, .versions li") + ] + + # Extract references + references = [ + a["href"] + for a in asset_entry.select(".references a, .external-links a") + if a.has_attr("href") + ] + + self.parsed_advisories.append( + { + "cve_id": cve_id, + "summary": description, + "severity": severity, + "affected_versions": affected_versions, + "references": references, + } + ) + except Exception as e: + self.log(f"Skipping invalid advisory block for {cve_id}: {str(e)}") + continue + + @transaction.atomic + def import_to_db(self): + if not hasattr(self, "parsed_advisories"): + self.log("No advisories to import.") + return + + for data in self.parsed_advisories: + vuln, _ = Vulnerability.objects.get_or_create( + vulnerability_id=data["cve_id"], + defaults={ + "summary": data["summary"], + "severity": data["severity"], + }, + ) + + # If references exist, create first reference + if data["references"]: + VulnerabilityReference.objects.get_or_create( + vulnerability=vuln, url=data["references"][0] + ) + + # Create AffectedPackage records + for version in data["affected_versions"]: + parsed_version = self.parse_versions(version) + purl_str = PackageURL( + type="liferay", + name="dxp" if "dxp" in parsed_version.lower() else "portal", + version=parsed_version, + ).to_string() + # Do something with purl_str, e.g., store in your DB or logs + self.log(f"Affected version PURL: {purl_str}") + + self.log(f"Imported {len(self.parsed_advisories)} Liferay advisories") + + def parse_versions(self, text): + if "DXP" in text: + return f"liferay-dxp-{text.split()[-1].lower()}" + return f"liferay-portal-{text.split()[0]}" From 0178fb5fbe790d562fb4a8e61a9bb9a1692fb9a6 Mon Sep 17 00:00:00 2001 From: Shrish0098 Date: Wed, 5 Mar 2025 10:38:59 +0530 Subject: [PATCH 2/3] Updated liferay_advisories Signed-off-by: Shrish0098 --- .../pipelines/liferay_advisories.py | 57 ++++++++++++++++++- 1 file changed, 56 insertions(+), 1 deletion(-) diff --git a/vulnerabilities/pipelines/liferay_advisories.py b/vulnerabilities/pipelines/liferay_advisories.py index 2c5a19f82..721d48328 100644 --- a/vulnerabilities/pipelines/liferay_advisories.py +++ b/vulnerabilities/pipelines/liferay_advisories.py @@ -1,3 +1,16 @@ +# +# 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. +# + + +from datetime import datetime +from typing import Iterable + import requests from bs4 import BeautifulSoup from django.db import transaction @@ -7,6 +20,8 @@ from requests.exceptions import RequestException from requests.exceptions import Timeout +from vulnerabilities.importer import AdvisoryData +from vulnerabilities.importer import Reference from vulnerabilities.models import Advisory from vulnerabilities.models import Vulnerability from vulnerabilities.models import VulnerabilityReference @@ -146,7 +161,7 @@ def import_to_db(self): name="dxp" if "dxp" in parsed_version.lower() else "portal", version=parsed_version, ).to_string() - # Do something with purl_str, e.g., store in your DB or logs + self.log(f"Affected version PURL: {purl_str}") self.log(f"Imported {len(self.parsed_advisories)} Liferay advisories") @@ -155,3 +170,43 @@ def parse_versions(self, text): if "DXP" in text: return f"liferay-dxp-{text.split()[-1].lower()}" return f"liferay-portal-{text.split()[0]}" + + def advisory_data(self) -> Iterable[AdvisoryData]: + import logging + + from vulnerabilities.importer import AffectedPackage + + logger = logging.getLogger(__name__) + + try: + # Execute pipeline steps + self.fetch_advisories() + self.parse_advisories() + + # Convert parsed data to AdvisoryData objects + for data in getattr(self, "parsed_advisories", []): + # Create affected packages list + affected_packages = [] + for version in data.get("affected_versions", []): + parsed_version = self.parse_versions(version) + affected_packages.append( + AffectedPackage( + package=PackageURL( + type="liferay", + name="dxp" if "dxp" in parsed_version.lower() else "portal", + version=parsed_version, + ) + ) + ) + + yield AdvisoryData( + aliases=[data["cve_id"]], + summary=data.get("summary", ""), + references=[Reference(url=ref) for ref in data.get("references", [])], + affected_packages=affected_packages, + severity=data.get("severity"), + ) + + except Exception as e: + logger.error(f"Error in Liferay pipeline: {str(e)}") + raise From ffd345798c32982f08f26f9f7c7c9d9a72603754 Mon Sep 17 00:00:00 2001 From: Shrish0098 Date: Fri, 7 Mar 2025 10:32:17 +0530 Subject: [PATCH 3/3] Improved Pipelines Signed-off-by: Shrish Mishra shrish409@gmail.com Signed-off-by: Shrish0098 --- .../pipelines/liferay_advisories.py | 253 ++++++++---------- 1 file changed, 114 insertions(+), 139 deletions(-) diff --git a/vulnerabilities/pipelines/liferay_advisories.py b/vulnerabilities/pipelines/liferay_advisories.py index 721d48328..e4fb3ad73 100644 --- a/vulnerabilities/pipelines/liferay_advisories.py +++ b/vulnerabilities/pipelines/liferay_advisories.py @@ -1,4 +1,3 @@ -# # Copyright (c) nexB Inc. and others. All rights reserved. # VulnerableCode is a trademark of nexB Inc. # SPDX-License-Identifier: Apache-2.0 @@ -7,206 +6,182 @@ # See https://aboutcode.org for more information about nexB OSS projects. # - -from datetime import datetime +import re from typing import Iterable import requests from bs4 import BeautifulSoup -from django.db import transaction -from fake_useragent import UserAgent 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.models import Advisory -from vulnerabilities.models import Vulnerability -from vulnerabilities.models import VulnerabilityReference -from vulnerabilities.pipelines import VulnerableCodePipeline +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline + +class LiferayAdvisoryPipeline(VulnerableCodeBaseImporterPipeline): + """Pipeline to import Liferay security advisories.""" -class LiferayAdvisoryPipeline(VulnerableCodePipeline): 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.import_to_db, + 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: - ua = UserAgent() response = requests.get( - "https://liferay.dev/portal/security/known-vulnerabilities", - headers={"User-Agent": ua.chrome}, + self.license_url, + headers={"User-Agent": "Mozilla/5.0"}, timeout=30, ) response.raise_for_status() - - if not response.text.strip(): - raise ValueError("Empty response from server") - self.html_content = response.text - - except Timeout as te: - self.log(f"Timeout occurred: {te}") - raise - except HTTPError as he: - self.log(f"HTTP Error {he.response.status_code}: {he}") - if he.response.status_code == 403: - self.log("Consider rotating User-Agent headers") - raise - except RequestException as re: - self.log(f"Request failed: {re}") + 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 RuntimeError(f"Fatal pipeline error: {e}") from e + raise def parse_advisories(self): + """Parse HTML to extract advisories.""" self.parsed_advisories = [] soup = BeautifulSoup(self.html_content, "html.parser") - portlet_section = soup.select_one("section.portlet") - if not portlet_section: - self.log("No portlet section found") + # 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 - # Loop over advisories - for asset_entry in portlet_section.select(".asset-entry, .entry, .asset-publisher"): - try: - cve_id = asset_entry.select_one("h1, h2, h3, .entry-title").get_text(strip=True) - except AttributeError: - self.log("No title found; skipping entry.") + # 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 - if not cve_id.startswith("CVE-"): - self.log(f"Skipping non-CVE entry: {cve_id}") + cve_id = match.group(1).upper() + if cve_id in processed_cves: continue - try: - description = asset_entry.select_one(".entry-content, .asset-content").get_text( - " ", strip=True - ) + processed_cves.add(cve_id) - metadata = asset_entry.select_one(".metadata, .entry-metadata") - if not metadata: - self.log(f"No metadata found for {cve_id}; skipping.") - continue + # 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 - # Attempt to parse severity and versions - severity_tag = metadata.select_one(".severity:contains('Severity') + dd") - severity = severity_tag.get_text(strip=True) if severity_tag else "Unknown" + if not container: + continue - affected_versions = [ - li.get_text(strip=True) - for li in metadata.select(".affected-versions li, .versions li") - ] + # 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) + ) - # Extract references - references = [ - a["href"] - for a in asset_entry.select(".references a, .external-links a") - if a.has_attr("href") - ] + 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, - "severity": severity, + "summary": description[:500] if description else "", "affected_versions": affected_versions, "references": references, } ) - except Exception as e: - self.log(f"Skipping invalid advisory block for {cve_id}: {str(e)}") - continue - - @transaction.atomic - def import_to_db(self): - if not hasattr(self, "parsed_advisories"): - self.log("No advisories to import.") - return - - for data in self.parsed_advisories: - vuln, _ = Vulnerability.objects.get_or_create( - vulnerability_id=data["cve_id"], - defaults={ - "summary": data["summary"], - "severity": data["severity"], - }, - ) - - # If references exist, create first reference - if data["references"]: - VulnerabilityReference.objects.get_or_create( - vulnerability=vuln, url=data["references"][0] - ) - - # Create AffectedPackage records - for version in data["affected_versions"]: - parsed_version = self.parse_versions(version) - purl_str = PackageURL( - type="liferay", - name="dxp" if "dxp" in parsed_version.lower() else "portal", - version=parsed_version, - ).to_string() - - self.log(f"Affected version PURL: {purl_str}") - self.log(f"Imported {len(self.parsed_advisories)} Liferay advisories") + self.log(f"Successfully parsed advisory for {cve_id}") - def parse_versions(self, text): - if "DXP" in text: - return f"liferay-dxp-{text.split()[-1].lower()}" - return f"liferay-portal-{text.split()[0]}" - - def advisory_data(self) -> Iterable[AdvisoryData]: - import logging - - from vulnerabilities.importer import AffectedPackage - - logger = logging.getLogger(__name__) - - try: - # Execute pipeline steps - self.fetch_advisories() - self.parse_advisories() - - # Convert parsed data to AdvisoryData objects - for data in getattr(self, "parsed_advisories", []): - # Create affected packages list - affected_packages = [] - for version in data.get("affected_versions", []): - parsed_version = self.parse_versions(version) - affected_packages.append( - AffectedPackage( - package=PackageURL( - type="liferay", - name="dxp" if "dxp" in parsed_version.lower() else "portal", - version=parsed_version, - ) - ) - ) + 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=[data["cve_id"]], - summary=data.get("summary", ""), - references=[Reference(url=ref) for ref in data.get("references", [])], + aliases=[advisory["cve_id"]], + summary=advisory["summary"], + references=[Reference(url=url) for url in advisory.get("references", [])], affected_packages=affected_packages, - severity=data.get("severity"), + url=self.license_url, ) - except Exception as e: - logger.error(f"Error in Liferay pipeline: {str(e)}") - raise + def advisories_count(self): + if hasattr(self, "parsed_advisories"): + return len(self.parsed_advisories) + return 0