|
| 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 | +import logging |
| 10 | +import re |
| 11 | +from pathlib import Path |
| 12 | +from typing import Iterable |
| 13 | + |
| 14 | +from fetchcode.vcs import fetch_via_vcs |
| 15 | + |
| 16 | +from vulnerabilities.importer import AdvisoryData |
| 17 | +from vulnerabilities.importer import ReferenceV2 |
| 18 | +from vulnerabilities.importer import VulnerabilitySeverity |
| 19 | +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 |
| 20 | +from vulnerabilities.severity_systems import GENERIC |
| 21 | +from vulnerabilities.utils import build_description |
| 22 | +from vulnerabilities.utils import create_weaknesses_list |
| 23 | +from vulnerabilities.utils import cwe_regex |
| 24 | +from vulnerabilities.utils import dedupe |
| 25 | +from vulnerabilities.utils import find_all_cve |
| 26 | +from vulnerabilities.utils import get_advisory_url |
| 27 | + |
| 28 | +logger = logging.getLogger(__name__) |
| 29 | + |
| 30 | + |
| 31 | +class FireeyeImporterPipeline(VulnerableCodeBaseImporterPipelineV2): |
| 32 | + spdx_license_expression = "CC-BY-SA-4.0 AND MIT" |
| 33 | + license_url = "https://github.com/mandiant/Vulnerability-Disclosures/blob/master/README.md" |
| 34 | + notice = """ |
| 35 | + Copyright (c) Mandiant |
| 36 | + The following licenses/licensing apply to this Mandiant repository: |
| 37 | + 1. CC BY-SA 4.0 - For CVE related information not including source code (such as PoCs) |
| 38 | + 2. MIT - For source code contained within provided CVE information |
| 39 | + """ |
| 40 | + repo_url = "git+https://github.com/mandiant/Vulnerability-Disclosures" |
| 41 | + pipeline_id = "fireeye_importer_v2" |
| 42 | + |
| 43 | + @classmethod |
| 44 | + def steps(cls): |
| 45 | + return ( |
| 46 | + cls.clone, |
| 47 | + cls.collect_and_store_advisories, |
| 48 | + cls.clean_downloads, |
| 49 | + ) |
| 50 | + |
| 51 | + def advisories_count(self): |
| 52 | + base_path = Path(self.vcs_response.dest_dir) |
| 53 | + return sum( |
| 54 | + 1 |
| 55 | + for p in base_path.glob("**/*") |
| 56 | + if p.suffix.lower() == ".md" or p.stem.upper() == "README" |
| 57 | + ) |
| 58 | + |
| 59 | + def clone(self): |
| 60 | + self.log(f"Cloning `{self.repo_url}`") |
| 61 | + self.vcs_response = fetch_via_vcs(self.repo_url) |
| 62 | + |
| 63 | + def collect_advisories(self) -> Iterable[AdvisoryData]: |
| 64 | + base_path = Path(self.vcs_response.dest_dir) |
| 65 | + for file_path in base_path.glob("**/*"): |
| 66 | + if file_path.suffix.lower() != ".md": |
| 67 | + continue |
| 68 | + |
| 69 | + if file_path.stem.upper() == "README": |
| 70 | + continue |
| 71 | + |
| 72 | + try: |
| 73 | + with open(file_path, encoding="utf-8-sig") as f: |
| 74 | + yield parse_advisory_data( |
| 75 | + raw_data=f.read(), file_path=file_path, base_path=base_path |
| 76 | + ) |
| 77 | + except UnicodeError: |
| 78 | + logger.error(f"Invalid File UnicodeError: {file_path}") |
| 79 | + |
| 80 | + def clean_downloads(self): |
| 81 | + if self.vcs_response: |
| 82 | + self.log(f"Removing cloned repository") |
| 83 | + self.vcs_response.delete() |
| 84 | + |
| 85 | + def on_failure(self): |
| 86 | + self.clean_downloads() |
| 87 | + |
| 88 | + |
| 89 | +def parse_advisory_data(raw_data, file_path, base_path) -> AdvisoryData: |
| 90 | + """ |
| 91 | + Parse a fireeye advisory repo and return an AdvisoryData or None. |
| 92 | + These files are in Markdown format. |
| 93 | + """ |
| 94 | + raw_data = raw_data.replace("\n\n", "\n") |
| 95 | + md_list = raw_data.split("\n") |
| 96 | + md_dict = md_list_to_dict(md_list) |
| 97 | + |
| 98 | + database_id = md_list[0][1::] |
| 99 | + summary = md_dict.get(database_id[1::]) or [] |
| 100 | + description = md_dict.get("## Description") or [] |
| 101 | + impact = md_dict.get("## Impact") |
| 102 | + cve_refs = md_dict.get("## CVE Reference") or [] |
| 103 | + cve_ids = md_dict.get("## CVE ID") or [] |
| 104 | + cleaned_cve_ids = [] |
| 105 | + for line in cve_ids: |
| 106 | + found_cves = find_all_cve(line) |
| 107 | + cleaned_cve_ids.extend(found_cves) |
| 108 | + |
| 109 | + references = md_dict.get("## References") or [] |
| 110 | + cwe_data = md_dict.get("## Common Weakness Enumeration") or [] |
| 111 | + |
| 112 | + advisory_id = file_path.stem |
| 113 | + aliases = dedupe([cve.strip() for cve in cleaned_cve_ids + cve_refs]) |
| 114 | + aliases = [aliase for aliase in aliases if aliase != advisory_id] |
| 115 | + advisory_url = get_advisory_url( |
| 116 | + file=file_path, |
| 117 | + base_path=base_path, |
| 118 | + url="https://github.com/mandiant/Vulnerability-Disclosures/blob/master/", |
| 119 | + ) |
| 120 | + |
| 121 | + return AdvisoryData( |
| 122 | + advisory_id=advisory_id, |
| 123 | + aliases=aliases, |
| 124 | + summary=build_description(" ".join(summary), " ".join(description)), |
| 125 | + references_v2=get_references(references), |
| 126 | + severities=get_severities(impact), |
| 127 | + weaknesses=get_weaknesses(cwe_data), |
| 128 | + url=advisory_url, |
| 129 | + original_advisory_text=raw_data, |
| 130 | + ) |
| 131 | + |
| 132 | + |
| 133 | +def get_references(references): |
| 134 | + """ |
| 135 | + Return a list of Reference from a list of URL reference in md format |
| 136 | + >>> get_references(["- http://1-4a.com/cgi-bin/alienform/af.cgi"]) |
| 137 | + [ReferenceV2(reference_id='', reference_type='', url='http://1-4a.com/cgi-bin/alienform/af.cgi')] |
| 138 | + >>> get_references(["- [Mitre CVE-2021-42712](https://www.cve.org/CVERecord?id=CVE-2021-42712)"]) |
| 139 | + [ReferenceV2(reference_id='', reference_type='', url='https://www.cve.org/CVERecord?id=CVE-2021-42712')] |
| 140 | + """ |
| 141 | + urls = [] |
| 142 | + for ref in references: |
| 143 | + clean_ref = ref.strip() |
| 144 | + clean_ref = clean_ref.lstrip("-* ") |
| 145 | + url = matcher_url(clean_ref) |
| 146 | + if url: |
| 147 | + urls.append(url) |
| 148 | + return [ReferenceV2(url=url) for url in urls if url] |
| 149 | + |
| 150 | + |
| 151 | +def matcher_url(ref) -> str: |
| 152 | + """ |
| 153 | + Returns URL of the reference markup from reference url in Markdown format |
| 154 | + """ |
| 155 | + markup_regex = "\[([^\[]+)]\(\s*(http[s]?://.+)\s*\)" |
| 156 | + matched_markup = re.findall(markup_regex, ref) |
| 157 | + if matched_markup: |
| 158 | + return matched_markup[0][1] |
| 159 | + else: |
| 160 | + return ref |
| 161 | + |
| 162 | + |
| 163 | +def md_list_to_dict(md_list): |
| 164 | + """ |
| 165 | + Returns a dictionary of md_list from a list of a md file splited by \n |
| 166 | + >>> md_list_to_dict(["# Header","hello" , "hello again" ,"# Header2"]) |
| 167 | + {'# Header': ['hello', 'hello again'], '# Header2': []} |
| 168 | + """ |
| 169 | + md_dict = {} |
| 170 | + md_key = "" |
| 171 | + for md_line in md_list: |
| 172 | + if md_line.startswith("#"): |
| 173 | + md_dict[md_line] = [] |
| 174 | + md_key = md_line |
| 175 | + else: |
| 176 | + md_dict[md_key].append(md_line) |
| 177 | + return md_dict |
| 178 | + |
| 179 | + |
| 180 | +def get_weaknesses(cwe_data): |
| 181 | + """ |
| 182 | + Return the list of CWE IDs as integers from a list of weakness summaries, e.g., [379]. |
| 183 | + >>> get_weaknesses([ |
| 184 | + ... "CWE-379: Creation of Temporary File in Directory with Insecure Permissions", |
| 185 | + ... "CWE-362: Concurrent Execution using Shared Resource with Improper Synchronization ('Race Condition')" |
| 186 | + ... ]) |
| 187 | + [379, 362] |
| 188 | + """ |
| 189 | + cwe_list = [] |
| 190 | + for line in cwe_data: |
| 191 | + cwe_ids = re.findall(cwe_regex, line) |
| 192 | + cwe_list.extend(cwe_ids) |
| 193 | + |
| 194 | + weaknesses = create_weaknesses_list(cwe_list) |
| 195 | + return weaknesses |
| 196 | + |
| 197 | + |
| 198 | +def get_severities(impact): |
| 199 | + """ |
| 200 | + Return a list of VulnerabilitySeverity extracted from the impact string. |
| 201 | + >>> get_severities([ |
| 202 | + ... "High - Arbitrary Ring 0 code execution", |
| 203 | + ... ]) |
| 204 | + [VulnerabilitySeverity(system=ScoringSystem(identifier='generic_textual', name='Generic textual severity rating', url='', notes='Severity for generic scoring systems. Contains generic textual values like High, Low etc'), value='High', scoring_elements='', published_at=None, url=None)] |
| 205 | + >>> get_severities([]) |
| 206 | + [] |
| 207 | + """ |
| 208 | + if not impact: |
| 209 | + return [] |
| 210 | + |
| 211 | + impact_text = impact[0] |
| 212 | + value = "" |
| 213 | + if " - " in impact_text: |
| 214 | + value = impact_text.split(" - ")[0] |
| 215 | + elif ": " in impact_text: |
| 216 | + value = impact_text.split(": ")[0] |
| 217 | + else: |
| 218 | + parts = impact_text.split(" ") |
| 219 | + if parts: |
| 220 | + value = parts[0] |
| 221 | + |
| 222 | + if not value.lower() in ["high", "medium", "low"]: |
| 223 | + return [] |
| 224 | + |
| 225 | + return [VulnerabilitySeverity(system=GENERIC, value=value)] |
0 commit comments