|
| 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 operator |
| 12 | +import re |
| 13 | +from itertools import groupby |
| 14 | +from pathlib import Path |
| 15 | + |
| 16 | +from fetchcode.vcs import fetch_via_vcs |
| 17 | +from packageurl import PackageURL |
| 18 | +from univers.version_range import NugetVersionRange |
| 19 | + |
| 20 | +from vulnerabilities.importer import AdvisoryData |
| 21 | +from vulnerabilities.importer import AffectedPackageV2 |
| 22 | +from vulnerabilities.importer import ReferenceV2 |
| 23 | +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 |
| 24 | +from vulnerabilities.utils import get_advisory_url |
| 25 | + |
| 26 | + |
| 27 | +class RetireDotnetImporterPipeline(VulnerableCodeBaseImporterPipelineV2): |
| 28 | + license_url = "https://github.com/RetireNet/Packages/blob/master/LICENSE" |
| 29 | + spdx_license_expression = "MIT" |
| 30 | + repo_url = "git+https://github.com/RetireNet/Packages/" |
| 31 | + pipeline_id = "retiredotnet_importer_v2" |
| 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 = "RetireNet-" + 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 | + packages = json_doc.get("packages") or [] |
| 68 | + key_func = operator.itemgetter("id") |
| 69 | + packages.sort(key=key_func) |
| 70 | + grouped_packages = groupby(packages, key=key_func) |
| 71 | + |
| 72 | + for key, group in grouped_packages: |
| 73 | + affected_versions = [] |
| 74 | + fixed_versions = [] |
| 75 | + |
| 76 | + for pkg in list(group): |
| 77 | + name = pkg.get("id") |
| 78 | + if not name: |
| 79 | + continue |
| 80 | + |
| 81 | + affected_version = pkg.get("affected") |
| 82 | + if affected_version: |
| 83 | + affected_versions.append(affected_version) |
| 84 | + |
| 85 | + fixed_version = pkg.get("fix") |
| 86 | + if fixed_version: |
| 87 | + fixed_versions.append(fixed_version) |
| 88 | + |
| 89 | + affected_version_range = None |
| 90 | + if affected_versions: |
| 91 | + affected_version_range = NugetVersionRange.from_versions(affected_versions) |
| 92 | + |
| 93 | + fixed_version_range = None |
| 94 | + if fixed_versions: |
| 95 | + fixed_version_range = NugetVersionRange.from_versions(affected_versions) |
| 96 | + |
| 97 | + if affected_packages: |
| 98 | + affected_packages.append( |
| 99 | + AffectedPackageV2( |
| 100 | + package=PackageURL(type="nuget", name=name), |
| 101 | + affected_version_range=affected_version_range, |
| 102 | + fixed_version_range=fixed_version_range, |
| 103 | + ) |
| 104 | + ) |
| 105 | + |
| 106 | + link = json_doc.get("link") |
| 107 | + if link: |
| 108 | + vuln_reference = [ |
| 109 | + ReferenceV2( |
| 110 | + url=link, |
| 111 | + ) |
| 112 | + ] |
| 113 | + |
| 114 | + yield AdvisoryData( |
| 115 | + advisory_id=advisory_id, |
| 116 | + aliases=[aliases] if aliases else [], |
| 117 | + summary=description, |
| 118 | + affected_packages=affected_packages, |
| 119 | + references_v2=vuln_reference, |
| 120 | + url=advisory_url, |
| 121 | + ) |
| 122 | + |
| 123 | + @staticmethod |
| 124 | + def vuln_id_from_desc(desc): |
| 125 | + cve_regex = re.compile(r"CVE-\d+-\d+") |
| 126 | + res = cve_regex.search(desc) |
| 127 | + if res: |
| 128 | + return desc[res.start() : res.end()] |
| 129 | + else: |
| 130 | + return None |
| 131 | + |
| 132 | + def clean_downloads(self): |
| 133 | + """Cleanup any temporary repository data.""" |
| 134 | + if self.vcs_response: |
| 135 | + self.log(f"Removing cloned repository") |
| 136 | + self.vcs_response.delete() |
| 137 | + |
| 138 | + def on_failure(self): |
| 139 | + """Ensure cleanup is always performed on failure.""" |
| 140 | + self.clean_downloads() |
0 commit comments