|
| 1 | +# |
| 2 | +# Copyright (c) nexB Inc. and others. All rights reserved. |
| 3 | +# VulnerableCode is a trademark of nexB Inc. |
| 4 | +# SPDX-License-Identifier: Apache-2.0 |
| 5 | +# See http://www.apache.org/licenses/LICENSE-2.0 for the license text. |
| 6 | +# See https://github.com/aboutcode-org/vulnerablecode for support or download. |
| 7 | +# See https://aboutcode.org for more information about nexB OSS projects. |
| 8 | +# |
| 9 | + |
| 10 | +import json |
| 11 | +import re |
| 12 | +from collections import defaultdict |
| 13 | +from pathlib import Path |
| 14 | + |
| 15 | +from fetchcode.vcs import fetch_via_vcs |
| 16 | +from packageurl import PackageURL |
| 17 | +from univers.version_range import NugetVersionRange |
| 18 | + |
| 19 | +from vulnerabilities.importer import AdvisoryDataV2 |
| 20 | +from vulnerabilities.importer import AffectedPackageV2 |
| 21 | +from vulnerabilities.importer import ReferenceV2 |
| 22 | +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 |
| 23 | +from vulnerabilities.utils import get_advisory_url |
| 24 | + |
| 25 | + |
| 26 | +class RetireDotnetImporterPipeline(VulnerableCodeBaseImporterPipelineV2): |
| 27 | + license_url = "https://github.com/RetireNet/Packages/blob/master/LICENSE" |
| 28 | + spdx_license_expression = "MIT" |
| 29 | + repo_url = "git+https://github.com/RetireNet/Packages/" |
| 30 | + pipeline_id = "retiredotnet_importer_v2" |
| 31 | + run_once = True |
| 32 | + |
| 33 | + @classmethod |
| 34 | + def steps(cls): |
| 35 | + return ( |
| 36 | + cls.clone, |
| 37 | + cls.collect_and_store_advisories, |
| 38 | + cls.clean_downloads, |
| 39 | + ) |
| 40 | + |
| 41 | + def clone(self): |
| 42 | + self.log(f"Cloning `{self.repo_url}`") |
| 43 | + self.vcs_response = fetch_via_vcs(self.repo_url) |
| 44 | + |
| 45 | + def advisories_count(self): |
| 46 | + root = Path(self.vcs_response.dest_dir) / "Content" |
| 47 | + return sum(1 for _ in root.rglob("*.json")) |
| 48 | + |
| 49 | + def collect_advisories(self): |
| 50 | + base_path = Path(self.vcs_response.dest_dir) |
| 51 | + vuln = base_path / "Content" |
| 52 | + affected_packages = [] |
| 53 | + |
| 54 | + for file in vuln.glob("*.json"): |
| 55 | + advisory_id = "retiredotnet-" + file.stem |
| 56 | + advisory_url = get_advisory_url( |
| 57 | + file=file, |
| 58 | + base_path=base_path, |
| 59 | + url="https://github.com/RetireNet/Packages/blob/master/", |
| 60 | + ) |
| 61 | + with open(file) as f: |
| 62 | + json_doc = json.load(f) |
| 63 | + description = json_doc.get("description") or "" |
| 64 | + aliases = self.vuln_id_from_desc(description) |
| 65 | + |
| 66 | + # group by package name `id` |
| 67 | + # { pkg_id: {'affected_versions': set(), 'fixed': set()} } |
| 68 | + grouped_packages = defaultdict( |
| 69 | + lambda: {"affected_versions": set(), "fixed_versions": set()} |
| 70 | + ) |
| 71 | + for pkg in json_doc.get("packages") or []: |
| 72 | + name = pkg.get("id") |
| 73 | + if not name: |
| 74 | + continue |
| 75 | + |
| 76 | + affected_version = pkg.get("affected") |
| 77 | + if affected_version: |
| 78 | + grouped_packages[name]["affected_versions"].add(affected_version) |
| 79 | + |
| 80 | + fixed_version = pkg.get("fix") |
| 81 | + if fixed_version: |
| 82 | + grouped_packages[name]["fixed_versions"].add(fixed_version) |
| 83 | + |
| 84 | + for pkg in grouped_packages: |
| 85 | + affected_version_range = None |
| 86 | + affected_versions = grouped_packages[pkg]["affected_versions"] |
| 87 | + if affected_versions: |
| 88 | + affected_version_range = NugetVersionRange.from_versions(affected_versions) |
| 89 | + |
| 90 | + fixed_version_range = None |
| 91 | + fixed_versions = grouped_packages[pkg]["fixed_versions"] |
| 92 | + if fixed_versions: |
| 93 | + fixed_version_range = NugetVersionRange.from_versions(fixed_versions) |
| 94 | + |
| 95 | + if affected_version_range or fixed_version_range: |
| 96 | + affected_packages.append( |
| 97 | + AffectedPackageV2( |
| 98 | + package=PackageURL(type="nuget", name=pkg), |
| 99 | + affected_version_range=affected_version_range, |
| 100 | + fixed_version_range=fixed_version_range, |
| 101 | + ) |
| 102 | + ) |
| 103 | + |
| 104 | + link = json_doc.get("link") |
| 105 | + if link: |
| 106 | + vuln_reference = [ |
| 107 | + ReferenceV2( |
| 108 | + url=link, |
| 109 | + ) |
| 110 | + ] |
| 111 | + |
| 112 | + yield AdvisoryDataV2( |
| 113 | + advisory_id=advisory_id, |
| 114 | + aliases=[aliases] if aliases else [], |
| 115 | + summary=description, |
| 116 | + affected_packages=affected_packages, |
| 117 | + references=vuln_reference, |
| 118 | + url=advisory_url, |
| 119 | + ) |
| 120 | + |
| 121 | + @staticmethod |
| 122 | + def vuln_id_from_desc(desc): |
| 123 | + cve_regex = re.compile(r"CVE-\d+-\d+") |
| 124 | + res = cve_regex.search(desc) |
| 125 | + if res: |
| 126 | + return desc[res.start() : res.end()] |
| 127 | + else: |
| 128 | + return None |
| 129 | + |
| 130 | + def clean_downloads(self): |
| 131 | + """Cleanup any temporary repository data.""" |
| 132 | + if self.vcs_response: |
| 133 | + self.log(f"Removing cloned repository") |
| 134 | + self.vcs_response.delete() |
| 135 | + |
| 136 | + def on_failure(self): |
| 137 | + """Ensure cleanup is always performed on failure.""" |
| 138 | + self.clean_downloads() |
0 commit comments