Add ArchLinux AdvisoryV2 importer pipeline

vulnerabilities/importers/__init__.py

@@ -42,6 +42,7 @@
 from vulnerabilities.pipelines import pypa_importer
 from vulnerabilities.pipelines import pysec_importer
 from vulnerabilities.pipelines.v2_importers import apache_httpd_importer as apache_httpd_v2
+from vulnerabilities.pipelines.v2_importers import archlinux_importer as archlinux_importer_v2
 from vulnerabilities.pipelines.v2_importers import (
     elixir_security_importer as elixir_security_importer_v2,
 )
@@ -99,5 +100,6 @@
         ubuntu_usn.UbuntuUSNImporter,
         fireeye.FireyeImporter,
         oss_fuzz.OSSFuzzImporter,
+        archlinux_importer_v2.ArchLinuxImporterPipeline,
     ]
 )

vulnerabilities/pipelines/v2_importers/archlinux_importer.py

@@ -0,0 +1,101 @@
+#
+# 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 typing import Iterable
+from typing import Mapping
+
+from packageurl import PackageURL
+from univers.version_range import ArchLinuxVersionRange
+from univers.versions import ArchLinuxVersion
+
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import ReferenceV2
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.utils import fetch_response
+
+
+class ArchLinuxImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
+    """ArchLinux Importer Pipeline"""
+
+    pipeline_id = "archlinux_importer_v2"
+    spdx_license_expression = "MIT"
+    license_url = "https://github.com/archlinux/arch-security-tracker/blob/master/LICENSE"
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.fetch,
+            cls.collect_and_store_advisories,
+        )
+
+    def fetch(self) -> Iterable[Mapping]:
+        url = "https://security.archlinux.org/json"
+        self.log(f"Fetching `{url}`")
+        response = fetch_response(url)
+        self.response = response.json()
+
+    def advisories_count(self) -> int:
+        return len(self.response)
+
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        for record in self.response:
+            yield self.parse_advisory(record)
+
+    def parse_advisory(self, record) -> AdvisoryData:
+        affected_packages = []
+        references = []
+        avg_name = record.get("name")
+        aliases = record.get("issues", [])
+        aliases.extend(record.get("advisories", []))
+        summary = record.get("type", "")
+        summary = "" if summary == "unknown" else summary
+
+        for name in record["packages"]:
+            affected = record.get("affected")
+            fixed = record.get("fixed")
+
+            affected_version_range = (
+                ArchLinuxVersionRange.from_versions([affected]) if affected else None
+            )
+            fixed_version = ArchLinuxVersion(fixed) if fixed else None
+            affected_package = AffectedPackage(
+                package=PackageURL(
+                    name=name,
+                    type="alpm",
+                    namespace="archlinux",
+                ),
+                affected_version_range=affected_version_range,
+                fixed_version=fixed_version,
+            )
+            affected_packages.append(affected_package)
+
+        references.append(
+            ReferenceV2(
+                reference_id=avg_name,
+                url="https://security.archlinux.org/{}".format(avg_name),
+            )
+        )
+        for ref in record["advisories"]:
+            references.append(
+                ReferenceV2(
+                    reference_id=ref,
+                    url="https://security.archlinux.org/{}".format(ref),
+                )
+            )
+
+        return AdvisoryData(
+            advisory_id=f"alpm/{avg_name}",
+            aliases=aliases,
+            summary=summary,
+            references_v2=references,
+            affected_packages=affected_packages,
+            weaknesses=[],
+            url=f"https://security.archlinux.org/{avg_name}.json",
+        )

vulnerabilities/tests/pipelines/v2_importers/test_archlinux_importer.py

@@ -0,0 +1,30 @@
+#
+# 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 json
+import os
+from pathlib import Path
+from unittest import TestCase
+
+from vulnerabilities.pipelines.v2_importers.archlinux_importer import ArchLinuxImporterPipeline
+from vulnerabilities.tests import util_tests
+
+TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "archlinux"
+
+
+class TestArchLinuxImporterPipeline(TestCase):
+    def test_to_advisories_with_summary(self):
+        archlinux_advisory_path = TEST_DATA / "archlinux-multi.json"
+
+        data = json.loads(archlinux_advisory_path.read_text(encoding="utf-8"))
+        expected_file = os.path.join(TEST_DATA, "archlinux_advisoryv2-expected.json")
+        pipeline = ArchLinuxImporterPipeline()
+        pipeline.response = data
+        result = [adv.to_dict() for adv in pipeline.collect_advisories()]
+        util_tests.check_results_against_json(result, expected_file)

vulnerabilities/tests/test_data/archlinux/archlinux_advisoryv2-expected.json

@@ -0,0 +1,77 @@
+[
+  {
+    "aliases": [
+      "CVE-2022-29217"
+    ],
+    "summary": "",
+    "affected_packages": [
+      {
+        "package": {
+          "type": "alpm",
+          "namespace": "archlinux",
+          "name": "python-pyjwt",
+          "version": "",
+          "qualifiers": "",
+          "subpath": ""
+        },
+        "affected_version_range": "vers:alpm/2.3.0-1",
+        "fixed_version": "2.4.0-1"
+      }
+    ],
+    "references": [],
+    "date_published": null,
+    "weaknesses": [],
+    "url": "https://security.archlinux.org/AVG-2781.json"
+  },
+  {
+    "aliases": [
+      "CVE-2022-26710",
+      "CVE-2022-22677",
+      "CVE-2022-22662"
+    ],
+    "summary": "",
+    "affected_packages": [
+      {
+        "package": {
+          "type": "alpm",
+          "namespace": "archlinux",
+          "name": "wpewebkit",
+          "version": "",
+          "qualifiers": "",
+          "subpath": ""
+        },
+        "affected_version_range": "vers:alpm/2.36.3-1",
+        "fixed_version": "2.36.4-1"
+      }
+    ],
+    "references": [],
+    "date_published": null,
+    "weaknesses": [],
+    "url": "https://security.archlinux.org/AVG-2780.json"
+  },
+  {
+    "aliases": [
+      "CVE-2016-3189",
+      "ASA-201702-19"
+    ],
+    "summary": "denial of service",
+    "affected_packages": [
+      {
+        "package": {
+          "type": "alpm",
+          "namespace": "archlinux",
+          "name": "bzip2",
+          "version": "",
+          "qualifiers": "",
+          "subpath": ""
+        },
+        "affected_version_range": "vers:alpm/1.0.6-5",
+        "fixed_version": "1.0.6-6"
+      }
+    ],
+    "references": [],
+    "date_published": null,
+    "weaknesses": [],
+    "url": "https://security.archlinux.org/AVG-4.json"
+  }
+]