Add advisory v2

vulnerabilities/forms.py

@@ -36,6 +36,14 @@ class VulnerabilitySearchForm(forms.Form):
     )
 
 
+class AdvisorySearchForm(forms.Form):
+
+    search = forms.CharField(
+        required=True,
+        widget=forms.TextInput(attrs={"placeholder": "Advisory id or alias such as CVE or GHSA"}),
+    )
+
+
 class ApiUserCreationForm(forms.ModelForm):
     """
     Support a simplified creation for API-only users directly from the UI.

vulnerabilities/importer.py

@@ -55,6 +55,7 @@ class VulnerabilitySeverity:
     value: str
     scoring_elements: str = ""
     published_at: Optional[datetime.datetime] = None
+    url: Optional[str] = None
 
     def to_dict(self):
         data = {
@@ -145,6 +146,54 @@ def from_url(cls, url):
         return cls(url=url)
 
 
+@dataclasses.dataclass(eq=True)
+@functools.total_ordering
+class ReferenceV2:
+    reference_id: str = ""
+    reference_type: str = ""
+    url: str = ""
+
+    def __post_init__(self):
+        if not self.url:
+            raise TypeError("Reference must have a url")
+        if self.reference_id and not isinstance(self.reference_id, str):
+            self.reference_id = str(self.reference_id)
+
+    def __lt__(self, other):
+        if not isinstance(other, Reference):
+            return NotImplemented
+        return self._cmp_key() < other._cmp_key()
+
+    # TODO: Add cache
+    def _cmp_key(self):
+        return (self.reference_id, self.reference_type, self.url)
+
+    def to_dict(self):
+        """Return a normalized dictionary representation"""
+        return {
+            "reference_id": self.reference_id,
+            "reference_type": self.reference_type,
+            "url": self.url,
+        }
+
+    @classmethod
+    def from_dict(cls, ref: dict):
+        return cls(
+            reference_id=str(ref["reference_id"]),
+            reference_type=ref.get("reference_type") or "",
+            url=ref["url"],
+        )
+
+    @classmethod
+    def from_url(cls, url):
+        reference_id = get_reference_id(url)
+        if "GHSA-" in reference_id.upper():
+            return cls(reference_id=reference_id, url=url)
+        if is_cve(reference_id):
+            return cls(url=url, reference_id=reference_id.upper())
+        return cls(url=url)
+
+
 class UnMergeablePackageError(Exception):
     """
     Raised when a package cannot be merged with another one.
@@ -298,10 +347,81 @@ class AdvisoryData:
     date_published must be aware datetime
     """
 
+    advisory_id: str = ""
     aliases: List[str] = dataclasses.field(default_factory=list)
     summary: Optional[str] = ""
     affected_packages: List[AffectedPackage] = dataclasses.field(default_factory=list)
     references: List[Reference] = dataclasses.field(default_factory=list)
+    references_v2: List[ReferenceV2] = dataclasses.field(default_factory=list)
+    date_published: Optional[datetime.datetime] = None
+    weaknesses: List[int] = dataclasses.field(default_factory=list)
+    severities: List[VulnerabilitySeverity] = dataclasses.field(default_factory=list)
+    url: Optional[str] = None
+
+    def __post_init__(self):
+        if self.date_published and not self.date_published.tzinfo:
+            logger.warning(f"AdvisoryData with no tzinfo: {self!r}")
+        if self.summary:
+            self.summary = self.clean_summary(self.summary)
+
+    def clean_summary(self, summary):
+        # https://nvd.nist.gov/vuln/detail/CVE-2013-4314
+        # https://github.com/cms-dev/cms/issues/888#issuecomment-516977572
+        summary = summary.strip()
+        if summary:
+            summary = summary.replace("\x00", "\uFFFD")
+        return summary
+
+    def to_dict(self):
+        return {
+            "aliases": self.aliases,
+            "summary": self.summary,
+            "affected_packages": [pkg.to_dict() for pkg in self.affected_packages],
+            "references": [ref.to_dict() for ref in self.references],
+            "date_published": self.date_published.isoformat() if self.date_published else None,
+            "weaknesses": self.weaknesses,
+            "url": self.url if self.url else "",
+        }
+
+    @classmethod
+    def from_dict(cls, advisory_data):
+        date_published = advisory_data["date_published"]
+        transformed = {
+            "aliases": advisory_data["aliases"],
+            "summary": advisory_data["summary"],
+            "affected_packages": [
+                AffectedPackage.from_dict(pkg)
+                for pkg in advisory_data["affected_packages"]
+                if pkg is not None
+            ],
+            "references": [Reference.from_dict(ref) for ref in advisory_data["references"]],
+            "date_published": datetime.datetime.fromisoformat(date_published)
+            if date_published
+            else None,
+            "weaknesses": advisory_data["weaknesses"],
+            "url": advisory_data.get("url") or None,
+        }
+        return cls(**transformed)
+
+
+@dataclasses.dataclass(order=True)
+class AdvisoryDataV2:
+    """
+    This data class expresses the contract between data sources and the import runner.
+
+    If a vulnerability_id is present then:
+        summary or affected_packages or references must be present
+    otherwise
+        either affected_package or references should be present
+
+    date_published must be aware datetime
+    """
+
+    advisory_id: str = ""
+    aliases: List[str] = dataclasses.field(default_factory=list)
+    summary: Optional[str] = ""
+    affected_packages: List[AffectedPackage] = dataclasses.field(default_factory=list)
+    references: List[ReferenceV2] = dataclasses.field(default_factory=list)
     date_published: Optional[datetime.datetime] = None
     weaknesses: List[int] = dataclasses.field(default_factory=list)
     url: Optional[str] = None

vulnerabilities/importers/__init__.py

@@ -34,6 +34,7 @@
 from vulnerabilities.importers import vulnrichment
 from vulnerabilities.importers import xen
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 from vulnerabilities.pipelines import alpine_linux_importer
 from vulnerabilities.pipelines import github_importer
 from vulnerabilities.pipelines import gitlab_importer
@@ -42,8 +43,24 @@
 from vulnerabilities.pipelines import nvd_importer
 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 github_importer as github_importer_v2
+from vulnerabilities.pipelines.v2_importers import gitlab_importer as gitlab_importer_v2
+from vulnerabilities.pipelines.v2_importers import npm_importer as npm_importer_v2
+from vulnerabilities.pipelines.v2_importers import nvd_importer as nvd_importer_v2
+from vulnerabilities.pipelines.v2_importers import pypa_importer as pypa_importer_v2
+from vulnerabilities.pipelines.v2_importers import pysec_importer as pysec_importer_v2
+from vulnerabilities.pipelines.v2_importers import vulnrichment_importer as vulnrichment_importer_v2
 
 IMPORTERS_REGISTRY = [
+    nvd_importer_v2.NVDImporterPipeline,
+    github_importer_v2.GitHubAPIImporterPipeline,
+    npm_importer_v2.NpmImporterPipeline,
+    vulnrichment_importer_v2.VulnrichImporterPipeline,
+    apache_httpd_v2.ApacheHTTPDImporterPipeline,
+    pypa_importer_v2.PyPaImporterPipeline,
+    gitlab_importer_v2.GitLabImporterPipeline,
+    pysec_importer_v2.PyPIImporterPipeline,
     nvd_importer.NVDImporterPipeline,
     github_importer.GitHubAPIImporterPipeline,
     gitlab_importer.GitLabImporterPipeline,
@@ -81,6 +98,9 @@
 ]
 
 IMPORTERS_REGISTRY = {
-    x.pipeline_id if issubclass(x, VulnerableCodeBaseImporterPipeline) else x.qualified_name: x
+    x.pipeline_id
+    if issubclass(x, VulnerableCodeBaseImporterPipeline)
+    or issubclass(x, VulnerableCodeBaseImporterPipelineV2)
+    else x.qualified_name: x
     for x in IMPORTERS_REGISTRY
 }

vulnerabilities/importers/curl.py

@@ -97,7 +97,7 @@ def parse_advisory_data(raw_data) -> AdvisoryData:
         ...     ]
         ... }
         >>> parse_advisory_data(raw_data)
-        AdvisoryData(aliases=['CVE-2024-2379'], summary='QUIC certificate check bypass with wolfSSL', affected_packages=[AffectedPackage(package=PackageURL(type='generic', namespace='curl.se', name='curl', version=None, qualifiers={}, subpath=None), affected_version_range=GenericVersionRange(constraints=(VersionConstraint(comparator='=', version=SemverVersion(string='8.6.0')),)), fixed_version=SemverVersion(string='8.7.0'))], references=[Reference(reference_id='', reference_type='', url='https://curl.se/docs/CVE-2024-2379.html', severities=[VulnerabilitySeverity(system=Cvssv3ScoringSystem(identifier='cvssv3.1', name='CVSSv3.1 Base Score', url='https://www.first.org/cvss/v3-1/', notes='CVSSv3.1 base score and vector'), value='Low', scoring_elements='', published_at=None)]), Reference(reference_id='', reference_type='', url='https://hackerone.com/reports/2410774', severities=[])], date_published=datetime.datetime(2024, 3, 27, 8, 0, tzinfo=datetime.timezone.utc), weaknesses=[297], url='https://curl.se/docs/CVE-2024-2379.json')
+        AdvisoryData(advisory_id='', aliases=['CVE-2024-2379'], summary='QUIC certificate check bypass with wolfSSL', affected_packages=[AffectedPackage(package=PackageURL(type='generic', namespace='curl.se', name='curl', version=None, qualifiers={}, subpath=None), affected_version_range=GenericVersionRange(constraints=(VersionConstraint(comparator='=', version=SemverVersion(string='8.6.0')),)), fixed_version=SemverVersion(string='8.7.0'))], references=[Reference(reference_id='', reference_type='', url='https://curl.se/docs/CVE-2024-2379.html', severities=[VulnerabilitySeverity(system=Cvssv3ScoringSystem(identifier='cvssv3.1', name='CVSSv3.1 Base Score', url='https://www.first.org/cvss/v3-1/', notes='CVSSv3.1 base score and vector'), value='Low', scoring_elements='', published_at=None, url=None)]), Reference(reference_id='', reference_type='', url='https://hackerone.com/reports/2410774', severities=[])], references_v2=[], date_published=datetime.datetime(2024, 3, 27, 8, 0, tzinfo=datetime.timezone.utc), weaknesses=[297], severities=[], url='https://curl.se/docs/CVE-2024-2379.json')
     """
 
     affected = get_item(raw_data, "affected")[0] if len(get_item(raw_data, "affected")) > 0 else []

vulnerabilities/importers/osv.py

@@ -107,6 +107,74 @@ def parse_advisory_data(
     )
 
 
+def parse_advisory_data_v2(
+    raw_data: dict, supported_ecosystems, advisory_url: str
+) -> Optional[AdvisoryData]:
+    """
+    Return an AdvisoryData build from a ``raw_data`` mapping of OSV advisory and
+    a ``supported_ecosystem`` string.
+    """
+    advisory_id = raw_data.get("id") or ""
+    if not advisory_id:
+        logger.error(f"Missing advisory id in OSV data: {raw_data}")
+        return None
+    summary = raw_data.get("summary") or ""
+    details = raw_data.get("details") or ""
+    summary = build_description(summary=summary, description=details)
+    aliases = raw_data.get("aliases") or []
+
+    date_published = get_published_date(raw_data=raw_data)
+    severities = list(get_severities(raw_data=raw_data))
+    references = get_references_v2(raw_data=raw_data)
+
+    affected_packages = []
+
+    for affected_pkg in raw_data.get("affected") or []:
+        purl = get_affected_purl(affected_pkg=affected_pkg, raw_id=advisory_id)
+
+        if not purl or purl.type not in supported_ecosystems:
+            logger.error(f"Unsupported package type: {affected_pkg!r} in OSV: {advisory_id!r}")
+            continue
+
+        affected_version_range = get_affected_version_range(
+            affected_pkg=affected_pkg,
+            raw_id=advisory_id,
+            supported_ecosystem=purl.type,
+        )
+
+        for fixed_range in affected_pkg.get("ranges") or []:
+            fixed_version = get_fixed_versions(
+                fixed_range=fixed_range, raw_id=advisory_id, supported_ecosystem=purl.type
+            )
+
+            for version in fixed_version:
+                affected_packages.append(
+                    AffectedPackage(
+                        package=purl,
+                        affected_version_range=affected_version_range,
+                        fixed_version=version,
+                    )
+                )
+    database_specific = raw_data.get("database_specific") or {}
+    cwe_ids = database_specific.get("cwe_ids") or []
+    weaknesses = list(map(get_cwe_id, cwe_ids))
+
+    if advisory_id in aliases:
+        aliases.remove(advisory_id)
+
+    return AdvisoryData(
+        advisory_id=advisory_id,
+        aliases=aliases,
+        summary=summary,
+        references_v2=references,
+        severities=severities,
+        affected_packages=affected_packages,
+        date_published=date_published,
+        weaknesses=weaknesses,
+        url=advisory_url,
+    )
+
+
 def extract_fixed_versions(fixed_range) -> Iterable[str]:
     """
     Return a list of fixed version strings given a ``fixed_range`` mapping of
@@ -187,6 +255,23 @@ def get_references(raw_data, severities) -> List[Reference]:
     return references
 
 
+def get_references_v2(raw_data) -> List[Reference]:
+    """
+    Return a list Reference extracted from a mapping of OSV ``raw_data`` given a
+    ``severities`` list of VulnerabilitySeverity.
+    """
+    references = []
+    for ref in raw_data.get("references") or []:
+        if not ref:
+            continue
+        url = ref["url"]
+        if not url:
+            logger.error(f"Reference without URL : {ref!r} for OSV id: {raw_data['id']!r}")
+            continue
+        references.append(Reference(url=ref["url"]))
+    return references
+
+
 def get_affected_purl(affected_pkg, raw_id):
     """
     Return an affected PackageURL or None given a mapping of ``affected_pkg``

vulnerabilities/improvers/__init__.py

@@ -20,6 +20,16 @@
 from vulnerabilities.pipelines import flag_ghost_packages
 from vulnerabilities.pipelines import populate_vulnerability_summary_pipeline
 from vulnerabilities.pipelines import remove_duplicate_advisories
+from vulnerabilities.pipelines.v2_improvers import compute_package_risk as compute_package_risk_v2
+from vulnerabilities.pipelines.v2_improvers import (
+    computer_package_version_rank as compute_version_rank_v2,
+)
+from vulnerabilities.pipelines.v2_improvers import enhance_with_exploitdb as exploitdb_v2
+from vulnerabilities.pipelines.v2_improvers import enhance_with_kev as enhance_with_kev_v2
+from vulnerabilities.pipelines.v2_improvers import (
+    enhance_with_metasploit as enhance_with_metasploit_v2,
+)
+from vulnerabilities.pipelines.v2_improvers import flag_ghost_packages as flag_ghost_packages_v2
 
 IMPROVERS_REGISTRY = [
     valid_versions.GitHubBasicImprover,
@@ -49,6 +59,12 @@
     add_cvss31_to_CVEs.CVEAdvisoryMappingPipeline,
     remove_duplicate_advisories.RemoveDuplicateAdvisoriesPipeline,
     populate_vulnerability_summary_pipeline.PopulateVulnerabilitySummariesPipeline,
+    exploitdb_v2.ExploitDBImproverPipeline,
+    enhance_with_kev_v2.VulnerabilityKevPipeline,
+    flag_ghost_packages_v2.FlagGhostPackagePipeline,
+    enhance_with_metasploit_v2.MetasploitImproverPipeline,
+    compute_package_risk_v2.ComputePackageRiskPipeline,
+    compute_version_rank_v2.ComputeVersionRankPipeline,
 ]
 
 IMPROVERS_REGISTRY = {

vulnerabilities/management/commands/import.py

@@ -14,6 +14,7 @@
 from vulnerabilities.import_runner import ImportRunner
 from vulnerabilities.importers import IMPORTERS_REGISTRY
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 
 
 class Command(BaseCommand):
@@ -57,7 +58,9 @@ def import_data(self, importers):
         failed_importers = []
 
         for importer in importers:
-            if issubclass(importer, VulnerableCodeBaseImporterPipeline):
+            if issubclass(importer, VulnerableCodeBaseImporterPipeline) or issubclass(
+                importer, VulnerableCodeBaseImporterPipelineV2
+            ):
                 self.stdout.write(f"Importing data using {importer.pipeline_id}")
                 status, error = importer().execute()
                 if status != 0: