|
| 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 | +from typing import Iterable |
| 11 | +from typing import NamedTuple |
| 12 | + |
| 13 | +import requests |
| 14 | +from bs4 import BeautifulSoup |
| 15 | +from packageurl import PackageURL |
| 16 | +from univers.version_range import NginxVersionRange |
| 17 | +from univers.versions import InvalidVersion |
| 18 | + |
| 19 | +from vulnerabilities.importer import AdvisoryData |
| 20 | +from vulnerabilities.importer import AffectedPackageV2 |
| 21 | +from vulnerabilities.importer import ReferenceV2 |
| 22 | +from vulnerabilities.importer import VulnerabilitySeverity |
| 23 | +from vulnerabilities.importer import logger |
| 24 | +from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline |
| 25 | +from vulnerabilities.severity_systems import GENERIC |
| 26 | + |
| 27 | + |
| 28 | +class NginxImporterPipeline(VulnerableCodeBaseImporterPipeline): |
| 29 | + """Collect Nginx security advisories.""" |
| 30 | + |
| 31 | + pipeline_id = "nginx_importer_v2" |
| 32 | + |
| 33 | + spdx_license_expression = "BSD-2-Clause" |
| 34 | + license_url = "https://nginx.org/LICENSE" |
| 35 | + url = "https://nginx.org/en/security_advisories.html" |
| 36 | + importer_name = "Nginx Importer" |
| 37 | + |
| 38 | + @classmethod |
| 39 | + def steps(cls): |
| 40 | + return ( |
| 41 | + cls.fetch, |
| 42 | + cls.collect_and_store_advisories, |
| 43 | + ) |
| 44 | + |
| 45 | + def fetch(self): |
| 46 | + self.log(f"Fetch `{self.url}`") |
| 47 | + self.advisory_data = requests.get(self.url).text |
| 48 | + |
| 49 | + def advisories_count(self): |
| 50 | + return self.advisory_data.count("<li><p>") |
| 51 | + |
| 52 | + def collect_advisories(self) -> Iterable[AdvisoryData]: |
| 53 | + """ |
| 54 | + Yield AdvisoryData from nginx security advisories HTML |
| 55 | + web page. |
| 56 | + """ |
| 57 | + soup = BeautifulSoup(self.advisory_data, features="lxml") |
| 58 | + vulnerability_list = soup.select("li p") |
| 59 | + for vulnerability_info in vulnerability_list: |
| 60 | + ngnix_advisory = parse_advisory_data_from_paragraph(vulnerability_info) |
| 61 | + yield to_advisory_data(ngnix_advisory) |
| 62 | + |
| 63 | + |
| 64 | +class NginxAdvisory(NamedTuple): |
| 65 | + advisory_id: str |
| 66 | + aliases: list |
| 67 | + summary: str |
| 68 | + severities: list |
| 69 | + not_vulnerable: str |
| 70 | + vulnerable: str |
| 71 | + references: list |
| 72 | + |
| 73 | + def to_dict(self): |
| 74 | + return self._asdict() |
| 75 | + |
| 76 | + |
| 77 | +def to_advisory_data(nginx_adv: NginxAdvisory) -> AdvisoryData: |
| 78 | + """ |
| 79 | + Return AdvisoryData from an NginxAdvisory tuple. |
| 80 | + """ |
| 81 | + package_name = "nginx" |
| 82 | + package_type = "nginx" |
| 83 | + qualifiers = {} |
| 84 | + |
| 85 | + purl = PackageURL(type=package_type, name=package_name, qualifiers=qualifiers) |
| 86 | + |
| 87 | + _, _, affected_versions = nginx_adv.vulnerable.partition(":") |
| 88 | + affected_versions = affected_versions.strip() |
| 89 | + |
| 90 | + if "nginx/Windows" in affected_versions: |
| 91 | + qualifiers["os"] = "windows" |
| 92 | + affected_versions = affected_versions.replace("nginx/Windows", "") |
| 93 | + |
| 94 | + _, _, fixed_versions = nginx_adv.not_vulnerable.partition(":") |
| 95 | + fixed_versions = fixed_versions.strip() |
| 96 | + |
| 97 | + if "nginx/Windows" in fixed_versions: |
| 98 | + qualifiers["os"] = "windows" |
| 99 | + fixed_versions = fixed_versions.replace("nginx/Windows", "") |
| 100 | + |
| 101 | + fixed_version_range = None |
| 102 | + try: |
| 103 | + fixed_version_range = NginxVersionRange.from_native(fixed_versions) |
| 104 | + except InvalidVersion: |
| 105 | + logger.error(f"Invalid vulnerable range {fixed_versions}") |
| 106 | + |
| 107 | + affected_version_range = None |
| 108 | + try: |
| 109 | + affected_version_range = NginxVersionRange.from_native(affected_versions) |
| 110 | + except InvalidVersion: |
| 111 | + logger.error(f"Invalid non vulnerable range {affected_versions}") |
| 112 | + |
| 113 | + affected_packages = [] |
| 114 | + if purl and affected_version_range or fixed_version_range: |
| 115 | + affected_packages.append( |
| 116 | + AffectedPackageV2( |
| 117 | + package=purl, |
| 118 | + affected_version_range=affected_version_range, |
| 119 | + fixed_version_range=fixed_version_range, |
| 120 | + ) |
| 121 | + ) |
| 122 | + |
| 123 | + return AdvisoryData( |
| 124 | + advisory_id=nginx_adv.advisory_id, |
| 125 | + aliases=nginx_adv.aliases, |
| 126 | + summary=nginx_adv.summary, |
| 127 | + affected_packages=affected_packages, |
| 128 | + references_v2=nginx_adv.references, |
| 129 | + url="https://nginx.org/en/security_advisories.html", |
| 130 | + ) |
| 131 | + |
| 132 | + |
| 133 | +def parse_advisory_data_from_paragraph(vulnerability_info): |
| 134 | + """ |
| 135 | + Return an NginxAdvisory from a ``vulnerability_info`` bs4 paragraph. |
| 136 | +
|
| 137 | + An advisory paragraph, without html markup, looks like this: |
| 138 | +
|
| 139 | + 1-byte memory overwrite in resolver |
| 140 | + Severity: medium |
| 141 | + Advisory |
| 142 | + CVE-2021-23017 |
| 143 | + Not vulnerable: 1.21.0+, 1.20.1+ |
| 144 | + Vulnerable: 0.6.18-1.20.0 |
| 145 | + The patch pgp |
| 146 | +
|
| 147 | + """ |
| 148 | + aliases = [] |
| 149 | + summary = None |
| 150 | + severities = [] |
| 151 | + not_vulnerable = None |
| 152 | + vulnerable = None |
| 153 | + references = [] |
| 154 | + is_first = True |
| 155 | + |
| 156 | + # we iterate on the children to accumulate values in variables |
| 157 | + # FIXME: using an explicit xpath-like query could be simpler |
| 158 | + for child in vulnerability_info.children: |
| 159 | + if is_first: |
| 160 | + summary = child |
| 161 | + is_first = False |
| 162 | + continue |
| 163 | + |
| 164 | + text = child.text.strip() |
| 165 | + text_low = text.lower() |
| 166 | + |
| 167 | + if text.startswith( |
| 168 | + ( |
| 169 | + "CVE-", |
| 170 | + "CORE-", |
| 171 | + "VU#", |
| 172 | + ) |
| 173 | + ): |
| 174 | + aliases.append(text) |
| 175 | + if text.startswith("CVE-"): |
| 176 | + # always keep the CVE as a reference too |
| 177 | + link = f"https://nvd.nist.gov/vuln/detail/{text}" |
| 178 | + reference = ReferenceV2(reference_id=text, url=link) |
| 179 | + references.append(reference) |
| 180 | + |
| 181 | + elif "severity" in text_low: |
| 182 | + severity = build_severity(severity=text) |
| 183 | + if severity: |
| 184 | + severities.append(severity) |
| 185 | + |
| 186 | + elif "not vulnerable" in text_low: |
| 187 | + not_vulnerable = text |
| 188 | + |
| 189 | + elif "vulnerable" in text_low: |
| 190 | + vulnerable = text |
| 191 | + |
| 192 | + elif hasattr(child, "attrs"): |
| 193 | + link = child.attrs.get("href") |
| 194 | + if link: |
| 195 | + if "cve.mitre.org" in link: |
| 196 | + references.append(ReferenceV2(reference_id=text, url=link)) |
| 197 | + elif "mailman.nginx.org" in link: |
| 198 | + references.append(ReferenceV2(url=link)) |
| 199 | + else: |
| 200 | + link = requests.compat.urljoin("https://nginx.org", link) |
| 201 | + references.append(ReferenceV2(url=link)) |
| 202 | + |
| 203 | + advisory_id = aliases.pop() |
| 204 | + return NginxAdvisory( |
| 205 | + advisory_id=advisory_id, |
| 206 | + aliases=aliases, |
| 207 | + summary=summary, |
| 208 | + severities=severities, |
| 209 | + not_vulnerable=not_vulnerable, |
| 210 | + vulnerable=vulnerable, |
| 211 | + references=references, |
| 212 | + ) |
| 213 | + |
| 214 | + |
| 215 | +def build_severity(severity): |
| 216 | + """ |
| 217 | + Return a VulnerabilitySeverity built from a ``severity`` string, or None. |
| 218 | +
|
| 219 | + For example:: |
| 220 | + >>> severity = "Severity: medium" |
| 221 | + >>> expected = VulnerabilitySeverity(system=GENERIC, value="medium") |
| 222 | + >>> assert build_severity(severity) == expected |
| 223 | + """ |
| 224 | + if severity.startswith("Severity:"): |
| 225 | + _, _, severity = severity.partition("Severity:") |
| 226 | + |
| 227 | + severity = severity.strip() |
| 228 | + if severity: |
| 229 | + return VulnerabilitySeverity(system=GENERIC, value=severity) |
0 commit comments