|
| 1 | +# |
| 2 | +# Copyright (c) nexB Inc. and others. All rights reserved. |
| 3 | +# SPDX-License-Identifier: Apache-2.0 |
| 4 | +# |
| 5 | + |
| 6 | +import json |
| 7 | +import logging |
| 8 | +from datetime import datetime |
| 9 | +from datetime import timezone |
| 10 | +from typing import Iterable |
| 11 | + |
| 12 | +from cwe2.database import Database |
| 13 | +from packageurl import PackageURL |
| 14 | +from univers.version_range import GenericVersionRange |
| 15 | +from univers.versions import SemverVersion |
| 16 | + |
| 17 | +from vulnerabilities.importer import AdvisoryData |
| 18 | +from vulnerabilities.importer import AffectedPackage |
| 19 | +from vulnerabilities.importer import ReferenceV2 |
| 20 | +from vulnerabilities.importer import VulnerabilitySeverity |
| 21 | +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 |
| 22 | +from vulnerabilities.severity_systems import SCORING_SYSTEMS |
| 23 | +from vulnerabilities.utils import fetch_response |
| 24 | +from vulnerabilities.utils import get_cwe_id |
| 25 | +from vulnerabilities.utils import get_item |
| 26 | + |
| 27 | +logger = logging.getLogger(__name__) |
| 28 | + |
| 29 | + |
| 30 | +class CurlImporterPipeline(VulnerableCodeBaseImporterPipelineV2): |
| 31 | + """ |
| 32 | + Pipeline-based importer for curl advisories from curl.se. |
| 33 | + """ |
| 34 | + |
| 35 | + pipeline_id = "curl_importer_v2" |
| 36 | + spdx_license_expression = "curl" |
| 37 | + license_url = "https://curl.se/docs/copyright.html" |
| 38 | + repo_url = "https://github.com/curl/curl-www/" |
| 39 | + url = "https://curl.se/docs/vuln.json" |
| 40 | + unfurl_version_ranges = True |
| 41 | + |
| 42 | + @classmethod |
| 43 | + def steps(cls): |
| 44 | + return (cls.collect_and_store_advisories,) |
| 45 | + |
| 46 | + def fetch_data(self): |
| 47 | + return fetch_response(self.url).json() |
| 48 | + |
| 49 | + def advisories_count(self) -> int: |
| 50 | + return len(self.fetch_data()) |
| 51 | + |
| 52 | + def collect_advisories(self) -> Iterable[AdvisoryData]: |
| 53 | + for entry in self.fetch_data(): |
| 54 | + cve_id = entry.get("aliases") or [] |
| 55 | + cve_id = cve_id[0] if cve_id else None |
| 56 | + if not cve_id or not cve_id.startswith("CVE"): |
| 57 | + package = get_item(entry, "database_specific", "package") |
| 58 | + logger.error(f"Invalid CVE ID: {cve_id} in package {package}") |
| 59 | + continue |
| 60 | + yield parse_curl_advisory(entry) |
| 61 | + |
| 62 | + |
| 63 | +def parse_curl_advisory(raw_data) -> AdvisoryData: |
| 64 | + """ |
| 65 | + Parse advisory data from raw JSON data and return an AdvisoryData object. |
| 66 | +
|
| 67 | + Args: |
| 68 | + raw_data (dict): Raw JSON data containing advisory information. |
| 69 | +
|
| 70 | + Returns: |
| 71 | + AdvisoryData: Parsed advisory data as an AdvisoryData object. |
| 72 | + """ |
| 73 | + affected = get_item(raw_data, "affected")[0] if len(get_item(raw_data, "affected")) > 0 else [] |
| 74 | + |
| 75 | + ranges = get_item(affected, "ranges")[0] if len(get_item(affected, "ranges")) > 0 else [] |
| 76 | + events = get_item(ranges, "events")[1] if len(get_item(ranges, "events")) > 1 else {} |
| 77 | + version_type = get_item(ranges, "type") if get_item(ranges, "type") else "" |
| 78 | + fixed_version = events.get("fixed") |
| 79 | + if version_type == "SEMVER" and fixed_version: |
| 80 | + fixed_version = SemverVersion(fixed_version) |
| 81 | + |
| 82 | + purl = PackageURL(type="generic", namespace="curl.se", name="curl") |
| 83 | + versions = affected.get("versions") or [] |
| 84 | + affected_version_range = GenericVersionRange.from_versions(versions) |
| 85 | + |
| 86 | + affected_package = AffectedPackage( |
| 87 | + package=purl, |
| 88 | + affected_version_range=affected_version_range, |
| 89 | + fixed_version=fixed_version, |
| 90 | + ) |
| 91 | + |
| 92 | + database_specific = raw_data.get("database_specific") or {} |
| 93 | + |
| 94 | + references = [] |
| 95 | + www_url = database_specific.get("www") |
| 96 | + issue_url = database_specific.get("issue") |
| 97 | + json_url = database_specific.get("URL") |
| 98 | + |
| 99 | + if www_url: |
| 100 | + references.append(ReferenceV2(url=www_url)) |
| 101 | + if issue_url: |
| 102 | + references.append(ReferenceV2(url=issue_url)) |
| 103 | + severity = VulnerabilitySeverity( |
| 104 | + system=SCORING_SYSTEMS["cvssv3.1"], value=database_specific.get("severity", ""), url=www_url |
| 105 | + ) |
| 106 | + |
| 107 | + published = raw_data.get("published", "") |
| 108 | + date_published = ( |
| 109 | + datetime.strptime(published, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc) |
| 110 | + if published |
| 111 | + else None |
| 112 | + ) |
| 113 | + |
| 114 | + weaknesses = get_cwe_from_curl_advisory(raw_data) |
| 115 | + |
| 116 | + aliases = raw_data.get("aliases", []) |
| 117 | + advisory_id = raw_data.get("id") or "" |
| 118 | + |
| 119 | + if advisory_id in aliases: |
| 120 | + aliases.remove(advisory_id) |
| 121 | + |
| 122 | + return AdvisoryData( |
| 123 | + advisory_id=advisory_id, |
| 124 | + aliases=aliases, |
| 125 | + summary=raw_data.get("summary") or "", |
| 126 | + affected_packages=[affected_package], |
| 127 | + references_v2=references, |
| 128 | + date_published=date_published, |
| 129 | + weaknesses=weaknesses, |
| 130 | + url=json_url, |
| 131 | + severities=[severity], |
| 132 | + original_advisory_text=json.dumps(raw_data, indent=2, ensure_ascii=False), |
| 133 | + ) |
| 134 | + |
| 135 | + |
| 136 | +def get_cwe_from_curl_advisory(raw_data): |
| 137 | + """ |
| 138 | + Extracts CWE IDs from the given raw_data and returns a list of CWE IDs. |
| 139 | +
|
| 140 | + >>> get_cwe_from_curl_advisory({"database_specific": {"CWE": {"id": "CWE-333"}}}) |
| 141 | + [333] |
| 142 | + >>> get_cwe_from_curl_advisory({"database_specific": {"CWE": {"id": ""}}}) |
| 143 | + [] |
| 144 | + """ |
| 145 | + weaknesses = [] |
| 146 | + db = Database() |
| 147 | + cwe_string = get_item(raw_data, "database_specific", "CWE", "id") or "" |
| 148 | + |
| 149 | + if cwe_string: |
| 150 | + try: |
| 151 | + cwe_id = get_cwe_id(cwe_string) |
| 152 | + db.get(cwe_id) # validate CWE exists |
| 153 | + weaknesses.append(cwe_id) |
| 154 | + except Exception: |
| 155 | + logger.error(f"Invalid CWE id: {cwe_string}") |
| 156 | + return weaknesses |
0 commit comments