Add ZyXEL v2 security advisories importer with tests

Signed-off-by: Tedsig42 <teddams047@gmail.com>

Author: Tednoob17

vulnerabilities/importers/__init__.py

@@ -84,6 +84,7 @@
 from vulnerabilities.pipelines.v2_importers import xen_importer as xen_importer_v2
 from vulnerabilities.utils import create_registry
 
+from vulnerabilities.pipelines.v2_importers import zyxel_importer as zyxel_importer_v2
 IMPORTERS_REGISTRY = create_registry(
     [
         archlinux_importer_v2.ArchLinuxImporterPipeline,
@@ -191,5 +192,6 @@
         collect_fix_commits_v2.CollectGitFixCommitsPipeline,
         collect_fix_commits_v2.CollectJenkinsFixCommitsPipeline,
         collect_fix_commits_v2.CollectGitlabFixCommitsPipeline,
+        zyxel_importer_v2.ZyxelImporterPipeline,
     ]
 )

vulnerabilities/pipelines/v2_importers/zyxel_importer.py

@@ -0,0 +1,211 @@
+#
+# 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 hashlib
+import logging
+import re
+from datetime import timezone
+from urllib.parse import urljoin
+from urllib.parse import urlparse
+
+import requests
+from bs4 import BeautifulSoup
+from dateutil import parser as date_parser
+
+from vulnerabilities.importer import AdvisoryDataV2
+from vulnerabilities.importer import ReferenceV2
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.utils import dedupe
+from vulnerabilities.utils import find_all_cve
+
+logger = logging.getLogger(__name__)
+
+
+class ZyxelImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
+    """Importer for ZyXEL security advisories pages."""
+
+    pipeline_id = "zyxel_importer_v2"
+    base_url = "https://www.zyxel.com/global/en/support/security-advisories"
+    spdx_license_expression = "NOASSERTION"
+    license_url = base_url
+
+    precedence = 200
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.fetch,
+            cls.collect_and_store_advisories,
+        )
+
+    def fetch(self):
+        self.log(f"Fetch `{self.base_url}`")
+        try:
+            response = requests.get(self.base_url, timeout=30)
+            response.raise_for_status()
+            self.listing_html = response.text
+        except requests.exceptions.Timeout:
+            self.log(f"Timeout while fetching {self.base_url}")
+            raise
+        except requests.exceptions.HTTPError as e:
+            self.log(f"HTTP error while fetching {self.base_url}: {e!r}")
+            raise
+        except requests.exceptions.RequestException as e:
+            self.log(f"Network error while fetching {self.base_url}: {e!r}")
+            raise
+
+    def advisories_count(self):
+        return len(parse_listing_for_advisory_urls(self.listing_html, self.base_url))
+
+    def collect_advisories(self):
+        for advisory_url in parse_listing_for_advisory_urls(self.listing_html, self.base_url):
+            try:
+                response = requests.get(advisory_url, timeout=30)
+                response.raise_for_status()
+                raw_html = response.text
+                advisory = parse_zyxel_advisory_page(raw_html=raw_html, advisory_url=advisory_url)
+                if advisory:
+                    yield advisory
+            except requests.exceptions.Timeout:
+                self.log(f"Timeout while fetching ZyXEL advisory at {advisory_url}")
+            except requests.exceptions.HTTPError as e:
+                self.log(f"HTTP error while fetching ZyXEL advisory at {advisory_url}: {e!r}")
+            except requests.exceptions.RequestException as e:
+                self.log(f"Network error while fetching ZyXEL advisory at {advisory_url}: {e!r}")
+            except Exception as e:
+                self.log(f"Unexpected error parsing ZyXEL advisory at {advisory_url}: {e!r}")
+
+
+def parse_listing_for_advisory_urls(raw_html, base_url):
+    """Return sorted advisory detail URLs from the ZyXEL listing page HTML."""
+    soup = BeautifulSoup(raw_html, features="lxml")
+    found_urls = set()
+
+    for anchor in soup.find_all("a", href=True):
+        href = anchor.get("href", "").strip()
+        if not href:
+            continue
+
+        absolute_url = urljoin(base_url, href)
+        parsed = urlparse(absolute_url)
+        slug = parsed.path.rstrip("/").split("/")[-1].lower()
+
+        if "support/security-advisories" not in absolute_url.lower():
+            continue
+
+        if slug == "security-advisories":
+            continue
+
+        found_urls.add(absolute_url)
+
+    return sorted(found_urls)
+
+
+def parse_zyxel_advisory_page(raw_html, advisory_url):
+    """Parse a ZyXEL advisory detail page and return AdvisoryDataV2."""
+    soup = BeautifulSoup(raw_html, features="lxml")
+    page_text = soup.get_text(" ", strip=True)
+
+    aliases = [alias.upper() for alias in find_all_cve(page_text)]
+    aliases = dedupe(aliases)
+
+    summary = extract_summary(soup=soup)
+    date_published = extract_published_date(soup=soup, page_text=page_text)
+    advisory_id = get_advisory_id(
+        advisory_url=advisory_url,
+        aliases=aliases,
+        summary=summary,
+        date_published=date_published,
+    )
+
+    references = get_references(soup=soup, advisory_url=advisory_url, aliases=aliases)
+
+    return AdvisoryDataV2(
+        advisory_id=advisory_id,
+        aliases=aliases,
+        summary=summary,
+        references=references,
+        date_published=date_published,
+        url=advisory_url,
+        original_advisory_text=raw_html,
+    )
+
+
+def extract_summary(soup):
+    h1 = soup.find("h1")
+    if h1 and h1.get_text(strip=True):
+        return h1.get_text(" ", strip=True)
+
+    title = soup.find("title")
+    if title and title.get_text(strip=True):
+        return title.get_text(" ", strip=True)
+
+    return "ZyXEL security advisory"
+
+
+def extract_published_date(soup, page_text):
+    for key, value in (
+        ("property", "article:published_time"),
+        ("name", "article:published_time"),
+        ("name", "publish_date"),
+        ("name", "date"),
+    ):
+        meta = soup.find("meta", attrs={key: value})
+        if not meta:
+            continue
+
+        content = (meta.get("content") or "").strip()
+        if not content:
+            continue
+
+        parsed = date_parser.parse(content)
+        if parsed:
+            if not parsed.tzinfo:
+                parsed = parsed.replace(tzinfo=timezone.utc)
+            return parsed
+
+    match = re.search(r"(?:published|release date)\s*:?\s*([A-Za-z0-9, :\-+/]+)", page_text, re.I)
+    if not match:
+        return None
+
+    parsed = date_parser.parse(match.group(1).strip())
+    if parsed and not parsed.tzinfo:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def get_advisory_id(advisory_url, aliases, summary, date_published):
+    slug = urlparse(advisory_url).path.rstrip("/").split("/")[-1]
+    if slug and slug.lower() != "security-advisories":
+        return f"zyxel-{slug}"
+
+    published = date_published.isoformat() if date_published else ""
+    digest = hashlib.sha1(
+        f"{advisory_url}|{summary}|{published}|{'|'.join(aliases)}".encode("utf-8")
+    ).hexdigest()[:16]
+    return f"zyxel-{digest}"
+
+
+def get_references(soup, advisory_url, aliases):
+    urls = [advisory_url]
+
+    for alias in aliases:
+        urls.append(f"https://nvd.nist.gov/vuln/detail/{alias}")
+
+    for anchor in soup.find_all("a", href=True):
+        href = anchor.get("href", "").strip()
+        if not href:
+            continue
+
+        absolute_url = urljoin(advisory_url, href)
+        if absolute_url.startswith("http"):
+            urls.append(absolute_url)
+
+    deduped_urls = dedupe(urls)
+    return [ReferenceV2(url=url) for url in deduped_urls]

vulnerabilities/tests/pipelines/v2_importers/test_zyxel_importer_pipeline.py

@@ -0,0 +1,61 @@
+#
+# 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 pathlib import Path
+
+from commoncode import testcase
+
+from vulnerabilities.pipelines.v2_importers import zyxel_importer
+
+
+class TestZyxelImporterPipeline(testcase.FileBasedTesting):
+    test_data_dir = Path(__file__).parent.parent.parent / "test_data" / "zyxel_v2"
+
+    def test_parse_listing_for_advisory_urls(self):
+        listing_file = self.get_test_loc("security_advisories_listing.html")
+        raw_html = Path(listing_file).read_text()
+
+        urls = zyxel_importer.parse_listing_for_advisory_urls(
+            raw_html=raw_html,
+            base_url="https://www.zyxel.com/global/en/support/security-advisories",
+        )
+
+        assert urls == [
+            "https://www.zyxel.com/global/en/support/security-advisories/zyxel-security-advisory-for-cve-2024-7261",
+            "https://www.zyxel.com/global/en/support/security-advisories/zyxel-security-advisory-for-cve-2024-7263",
+        ]
+
+    def test_parse_zyxel_advisory_page_extracts_cves_and_id(self):
+        advisory_file = self.get_test_loc("zyxel_security_advisory_for_foo.html")
+        raw_html = Path(advisory_file).read_text()
+
+        result = zyxel_importer.parse_zyxel_advisory_page(
+            raw_html=raw_html,
+            advisory_url="https://www.zyxel.com/global/en/support/security-advisories/zyxel-security-advisory-for-foo",
+        )
+
+        assert result.advisory_id == "zyxel-zyxel-security-advisory-for-foo"
+        assert result.summary == "ZyXEL Security Advisory for Foo"
+        assert result.aliases == ["CVE-2025-12345", "CVE-2025-67890"]
+        assert result.date_published.isoformat() == "2025-03-10T00:00:00+00:00"
+
+        reference_urls = [ref.url for ref in result.references]
+        assert "https://nvd.nist.gov/vuln/detail/CVE-2025-12345" in reference_urls
+        assert "https://nvd.nist.gov/vuln/detail/CVE-2025-67890" in reference_urls
+
+    def test_get_advisory_id_hash_fallback_when_slug_missing(self):
+        advisory_id = zyxel_importer.get_advisory_id(
+            advisory_url="https://www.zyxel.com/global/en/support/security-advisories/",
+            aliases=["CVE-2025-12345"],
+            summary="Example advisory",
+            date_published=None,
+        )
+
+        assert advisory_id.startswith("zyxel-")
+        assert advisory_id != "zyxel-security-advisories"

vulnerabilities/tests/test_data/zyxel_v2/security_advisories_listing.html

@@ -0,0 +1,7 @@
+<html>
+  <body>
+    <a href="https://www.zyxel.com/global/en/support/security-advisories">Listing</a>
+    <a href="https://www.zyxel.com/global/en/support/security-advisories/zyxel-security-advisory-for-cve-2024-7261">Advisory 1</a>
+    <a href="https://www.zyxel.com/global/en/support/security-advisories/zyxel-security-advisory-for-cve-2024-7263">Advisory 2</a>
+  </body>
+</html>

vulnerabilities/tests/test_data/zyxel_v2/zyxel_security_advisory_for_foo.html

@@ -0,0 +1,11 @@
+<html>
+  <head>
+    <title>ZyXEL Security Advisory for Foo</title>
+    <meta property="article:published_time" content="2025-03-10" />
+  </head>
+  <body>
+    <h1>ZyXEL Security Advisory for Foo</h1>
+    <p>This advisory addresses CVE-2025-12345 and CVE-2025-67890 in affected devices.</p>
+    <a href="https://www.cve.org/CVERecord?id=CVE-2025-12345">CVE Link</a>
+  </body>
+</html>