aboutcode-org
diff --git a/‎vulnerabilities/api_v2.py
Lines changed: 382 additions & 0 deletions b/‎vulnerabilities/api_v2.py
Lines changed: 382 additions & 0 deletions
diff --git a/‎vulnerabilities/forms.py
Lines changed: 8 additions & 0 deletions b/‎vulnerabilities/forms.py
Lines changed: 8 additions & 0 deletions
diff --git a/‎vulnerabilities/importer.py
Lines changed: 120 additions & 0 deletions b/‎vulnerabilities/importer.py
Lines changed: 120 additions & 0 deletions
diff --git a/‎vulnerabilities/importers/__init__.py
Lines changed: 55 additions & 42 deletions b/‎vulnerabilities/importers/__init__.py
Lines changed: 55 additions & 42 deletions
diff --git a/‎vulnerabilities/importers/curl.py
Lines changed: 1 addition & 1 deletion b/‎vulnerabilities/importers/curl.py
Lines changed: 1 addition & 1 deletion
diff --git a/‎vulnerabilities/importers/osv.py
Lines changed: 85 additions & 0 deletions b/‎vulnerabilities/importers/osv.py
Lines changed: 85 additions & 0 deletions
@@ -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.
 
@@ -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.
@@ -302,10 +351,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
 
@@ -33,7 +33,6 @@
 from vulnerabilities.importers import ubuntu_usn
 from vulnerabilities.importers import vulnrichment
 from vulnerabilities.importers import xen
-from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
 from vulnerabilities.pipelines import alpine_linux_importer
 from vulnerabilities.pipelines import github_importer
 from vulnerabilities.pipelines import gitlab_importer
@@ -42,45 +41,59 @@
 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
+from vulnerabilities.utils import create_registry
 
-IMPORTERS_REGISTRY = [
-    nvd_importer.NVDImporterPipeline,
-    github_importer.GitHubAPIImporterPipeline,
-    gitlab_importer.GitLabImporterPipeline,
-    github_osv.GithubOSVImporter,
-    pypa_importer.PyPaImporterPipeline,
-    npm_importer.NpmImporterPipeline,
-    nginx_importer.NginxImporterPipeline,
-    pysec_importer.PyPIImporterPipeline,
-    apache_tomcat.ApacheTomcatImporter,
-    postgresql.PostgreSQLImporter,
-    debian.DebianImporter,
-    curl.CurlImporter,
-    epss.EPSSImporter,
-    vulnrichment.VulnrichImporter,
-    alpine_linux_importer.AlpineLinuxImporterPipeline,
-    ruby.RubyImporter,
-    apache_kafka.ApacheKafkaImporter,
-    openssl.OpensslImporter,
-    redhat.RedhatImporter,
-    archlinux.ArchlinuxImporter,
-    ubuntu.UbuntuImporter,
-    debian_oval.DebianOvalImporter,
-    retiredotnet.RetireDotnetImporter,
-    apache_httpd.ApacheHTTPDImporter,
-    mozilla.MozillaImporter,
-    gentoo.GentooImporter,
-    istio.IstioImporter,
-    project_kb_msr2019.ProjectKBMSRImporter,
-    suse_scores.SUSESeverityScoreImporter,
-    elixir_security.ElixirSecurityImporter,
-    xen.XenImporter,
-    ubuntu_usn.UbuntuUSNImporter,
-    fireeye.FireyeImporter,
-    oss_fuzz.OSSFuzzImporter,
-]
-
-IMPORTERS_REGISTRY = {
-    x.pipeline_id if issubclass(x, VulnerableCodeBaseImporterPipeline) else x.qualified_name: x
-    for x in IMPORTERS_REGISTRY
-}
+IMPORTERS_REGISTRY = create_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,
+        github_osv.GithubOSVImporter,
+        pypa_importer.PyPaImporterPipeline,
+        npm_importer.NpmImporterPipeline,
+        nginx_importer.NginxImporterPipeline,
+        pysec_importer.PyPIImporterPipeline,
+        apache_tomcat.ApacheTomcatImporter,
+        postgresql.PostgreSQLImporter,
+        debian.DebianImporter,
+        curl.CurlImporter,
+        epss.EPSSImporter,
+        vulnrichment.VulnrichImporter,
+        alpine_linux_importer.AlpineLinuxImporterPipeline,
+        ruby.RubyImporter,
+        apache_kafka.ApacheKafkaImporter,
+        openssl.OpensslImporter,
+        redhat.RedhatImporter,
+        archlinux.ArchlinuxImporter,
+        ubuntu.UbuntuImporter,
+        debian_oval.DebianOvalImporter,
+        retiredotnet.RetireDotnetImporter,
+        apache_httpd.ApacheHTTPDImporter,
+        mozilla.MozillaImporter,
+        gentoo.GentooImporter,
+        istio.IstioImporter,
+        project_kb_msr2019.ProjectKBMSRImporter,
+        suse_scores.SUSESeverityScoreImporter,
+        elixir_security.ElixirSecurityImporter,
+        xen.XenImporter,
+        ubuntu_usn.UbuntuUSNImporter,
+        fireeye.FireyeImporter,
+        oss_fuzz.OSSFuzzImporter,
+    ]
+)
@@ -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 []
 
@@ -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``