|
| 1 | +import asyncio |
| 2 | +import re |
| 3 | +from typing import List |
| 4 | +from typing import Tuple |
| 5 | + |
| 6 | +import requests |
| 7 | +from bs4 import BeautifulSoup |
| 8 | +from dephell_specifier import RangeSpecifier |
| 9 | +from packageurl import PackageURL |
| 10 | + |
| 11 | +from vulnerabilities.data_source import Advisory |
| 12 | +from vulnerabilities.data_source import DataSource |
| 13 | +from vulnerabilities.data_source import Reference |
| 14 | +from vulnerabilities.data_source import VulnerabilitySeverity |
| 15 | +from vulnerabilities.package_managers import GitHubTagsAPI |
| 16 | +from vulnerabilities.severity_systems import scoring_systems |
| 17 | + |
| 18 | +SECURITY_UPDATES_URL = "https://mattermost.com/security-updates" |
| 19 | +MM_REPO = { |
| 20 | + "Mattermost Mobile Apps": "mattermost/mattermost-mobile", |
| 21 | + "Mattermost Server": "mattermost/mattermost-server", |
| 22 | + "Mattermost Desktop App": "mattermost/desktop", |
| 23 | +} |
| 24 | + |
| 25 | + |
| 26 | +class MattermostDataSource(DataSource): |
| 27 | + def updated_advisories(self): |
| 28 | + # FIXME: Change after this https://forum.mattermost.org/t/mattermost-website-returning-403-when-headers-contain-the-word-python/11412 |
| 29 | + self.set_api() |
| 30 | + data = requests.get( |
| 31 | + SECURITY_UPDATES_URL, headers={"user-agent": "aboutcode/vulnerablecode"} |
| 32 | + ).content |
| 33 | + return self.batch_advisories(self.to_advisories(data)) |
| 34 | + |
| 35 | + def set_api(self): |
| 36 | + self.version_api = GitHubTagsAPI() |
| 37 | + asyncio.run( |
| 38 | + self.version_api.load_api( |
| 39 | + [ |
| 40 | + MM_REPO["Mattermost Mobile Apps"], |
| 41 | + MM_REPO["Mattermost Server"], |
| 42 | + MM_REPO["Mattermost Desktop App"], |
| 43 | + ] |
| 44 | + ) |
| 45 | + ) |
| 46 | + |
| 47 | + def to_advisories(self, data): |
| 48 | + advisories = [] |
| 49 | + soup = BeautifulSoup(data, features="lxml") |
| 50 | + for row in soup.table.tbody.find_all("tr"): |
| 51 | + ( |
| 52 | + ref_col, |
| 53 | + severity_col, |
| 54 | + affected_col, |
| 55 | + _, |
| 56 | + fixed_col, |
| 57 | + desc_col, |
| 58 | + name_col, |
| 59 | + ) = row.select("td") |
| 60 | + |
| 61 | + name = name_col.text.strip() |
| 62 | + if name not in MM_REPO: |
| 63 | + continue |
| 64 | + |
| 65 | + fixed_versions = split_versions(fixed_col.text) |
| 66 | + fixed_packages = [ |
| 67 | + PackageURL( |
| 68 | + type="mattermost", |
| 69 | + name=name, |
| 70 | + version=version, |
| 71 | + ) |
| 72 | + for version in fixed_versions |
| 73 | + ] |
| 74 | + |
| 75 | + ( |
| 76 | + affected_version_ranges, |
| 77 | + excluded_version_ranges, |
| 78 | + ) = to_affected_version_ranges(affected_col.text, fixed_col.text) |
| 79 | + |
| 80 | + affected_packages = [ |
| 81 | + PackageURL(type="mattermost", name=name, version=version) |
| 82 | + for version in self.version_api.get(MM_REPO[name]) |
| 83 | + if |
| 84 | + # The versions comparisions and advisories are not compatible with cloud-* versions |
| 85 | + not version.startswith("cloud-") |
| 86 | + and any((version in version_range for version_range in affected_version_ranges)) |
| 87 | + and not any((version in version_range for version_range in excluded_version_ranges)) |
| 88 | + ] |
| 89 | + |
| 90 | + # Severities are either "na" or cvssv3.1_qr |
| 91 | + references = [ |
| 92 | + Reference( |
| 93 | + reference_id=ref_col.text, |
| 94 | + url=SECURITY_UPDATES_URL, |
| 95 | + severities=[ |
| 96 | + VulnerabilitySeverity( |
| 97 | + system=scoring_systems["cvssv3.1_qr"], value=severity_col.text |
| 98 | + ) |
| 99 | + ] |
| 100 | + if severity_col.text.lower() != "na" |
| 101 | + else [], |
| 102 | + ) |
| 103 | + ] |
| 104 | + |
| 105 | + for cve_id in re.findall(r"cve-\d+-\d+", desc_col.text, re.IGNORECASE): |
| 106 | + references.append( |
| 107 | + Reference( |
| 108 | + reference_id=cve_id, |
| 109 | + url=f"https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve_id}", |
| 110 | + ) |
| 111 | + ) |
| 112 | + advisories.append( |
| 113 | + Advisory( |
| 114 | + vulnerability_id="", |
| 115 | + summary=desc_col.text, |
| 116 | + references=references, |
| 117 | + impacted_package_urls=affected_packages, |
| 118 | + resolved_package_urls=fixed_packages, |
| 119 | + ) |
| 120 | + ) |
| 121 | + return advisories |
| 122 | + |
| 123 | + |
| 124 | +def split_versions(versions: str) -> List[str]: |
| 125 | + """ |
| 126 | + The versions can take the form: |
| 127 | + - v1.2,2.2 and 3.2 -> [1.2,2.2,3.2] |
| 128 | + - v1 and v2 -> [1,2] |
| 129 | + - v1, v2 -> [1,2] |
| 130 | + - <10 -> [<10] |
| 131 | + - na -> [] |
| 132 | + - all -> all (see `affected_version_ranges`) |
| 133 | + Returns list of versions without leading 'v' |
| 134 | + """ |
| 135 | + versions = versions.lower().strip().replace("and", ",") |
| 136 | + if versions == "na": |
| 137 | + return [] |
| 138 | + if versions == "all": |
| 139 | + return ["all"] |
| 140 | + |
| 141 | + versions = [ |
| 142 | + # some versions are like v2.4, remove v |
| 143 | + version.strip().replace("v", "") |
| 144 | + for version in versions.split(",") |
| 145 | + if version.strip() |
| 146 | + ] |
| 147 | + return versions |
| 148 | + |
| 149 | + |
| 150 | +def to_affected_version_ranges( |
| 151 | + affected_col: str, fixed_col: str |
| 152 | +) -> Tuple[List[RangeSpecifier], List[RangeSpecifier]]: |
| 153 | + """ |
| 154 | + affected_col could be of type "v5.20.x to v5.26.x, excluding v5.25.5 and v5.26.2" |
| 155 | + fixed_col is only relevent in case affected_col is "all" |
| 156 | + "all" means all the versions before the only present fixed. If there are many fixed versions, it doesn't return anything. |
| 157 | + Needs to be improved after https://github.com/nexB/vulnerablecode/issues/119 |
| 158 | + According to https://forum.mattermost.org/t/all-affected-versions-in-the-mattermost-advisory/11423, |
| 159 | +
|
| 160 | + Returns affected version included_ranges, excluded_ranges |
| 161 | + """ |
| 162 | + fixed_versions = split_versions(fixed_col) |
| 163 | + affected_col = affected_col.replace(".x", ".*") # For 5.20.x |
| 164 | + included, *excluded = affected_col.split("excluding") |
| 165 | + range_expressions = split_versions(included) |
| 166 | + |
| 167 | + if len(range_expressions) == 1: |
| 168 | + # special cases |
| 169 | + if range_expressions[0] == "na": |
| 170 | + return [], [] |
| 171 | + |
| 172 | + if range_expressions[0] == "all": |
| 173 | + if len(fixed_versions) > 1: |
| 174 | + # it gets very complicated. see link above |
| 175 | + return [], [RangeSpecifier()] |
| 176 | + return [RangeSpecifier(f"<{fixed_versions[0]}")], [] |
| 177 | + |
| 178 | + included_ranges = [] |
| 179 | + for range_expression in range_expressions: |
| 180 | + if "to" in range_expression: |
| 181 | + # eg range_expression == "3.2.0 to 3.2.1" |
| 182 | + lower_bound, upper_bound = range_expression.split("to") |
| 183 | + lower_bound = f">={lower_bound}" |
| 184 | + upper_bound = f"<={upper_bound}" |
| 185 | + included_ranges.append(RangeSpecifier(f"{lower_bound},{upper_bound}")) |
| 186 | + else: |
| 187 | + included_ranges.append(RangeSpecifier(range_expression)) |
| 188 | + |
| 189 | + excluded_ranges = [] |
| 190 | + if len(excluded): |
| 191 | + excluded_ranges = [RangeSpecifier(v) for v in split_versions(excluded[0])] |
| 192 | + |
| 193 | + return included_ranges, excluded_ranges |
0 commit comments