|
8 | 8 | # |
9 | 9 |
|
10 | 10 | import json |
| 11 | +import logging |
| 12 | +from io import BytesIO |
11 | 13 | from typing import Iterable |
| 14 | +from zipfile import ZipFile |
12 | 15 |
|
13 | | -from dateutil.parser import parse |
14 | | -from packageurl import PackageURL |
15 | | -from pytz import UTC |
16 | | -from univers.version_range import RANGE_CLASS_BY_SCHEMES |
| 16 | +import requests |
17 | 17 |
|
18 | 18 | from vulnerabilities.importer import AdvisoryDataV2 |
19 | | -from vulnerabilities.importer import AffectedPackageV2 |
20 | | -from vulnerabilities.importer import VulnerabilitySeverity |
21 | 19 | from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2 |
22 | | -from vulnerabilities.severity_systems import GENERIC |
23 | | -from vulnerabilities.utils import fetch_response |
24 | | - |
25 | | -# See https://docs.tuxcare.com/els-for-os/#cve-status-definition |
26 | | -NON_AFFECTED_STATUSES = ["Not Vulnerable"] |
27 | | -AFFECTED_STATUSES = ["Ignored", "Needs Triage", "In Testing", "In Progress", "In Rollout"] |
28 | | -FIXED_STATUSES = ["Released", "Already Fixed"] |
| 20 | +from vulnerabilities.pipes.osv_v2 import parse_advisory_data_v3 |
29 | 21 |
|
30 | 22 |
|
31 | 23 | class TuxCareImporterPipeline(VulnerableCodeBaseImporterPipelineV2): |
| 24 | + """ |
| 25 | + TuxCare Importer Pipeline |
| 26 | + Collect advisories from TuxCare OSV feed. |
| 27 | + """ |
| 28 | + |
32 | 29 | pipeline_id = "tuxcare_importer_v2" |
33 | | - spdx_license_expression = "Apache-2.0" |
| 30 | + datasource_id = "tuxcare" |
34 | 31 | license_url = "https://tuxcare.com/legal" |
35 | | - |
| 32 | + spdx_license_expression = "Apache-2.0" |
| 33 | + url = "https://osv-vulnerabilities.storage.googleapis.com/TuxCare/all.zip" |
36 | 34 | precedence = 100 |
37 | 35 |
|
38 | 36 | @classmethod |
39 | 37 | def steps(cls): |
40 | 38 | return ( |
41 | | - cls.fetch, |
42 | | - cls.group_records_by_cve, |
| 39 | + cls.fetch_zip, |
43 | 40 | cls.collect_and_store_advisories, |
44 | 41 | ) |
45 | 42 |
|
46 | | - def fetch(self) -> None: |
47 | | - url = "https://cve.tuxcare.com/els/download-json?orderBy=updated-desc" |
48 | | - self.log(f"Fetching `{url}`") |
49 | | - response = fetch_response(url) |
50 | | - self.response = response.json() if response else [] |
51 | | - |
52 | | - def group_records_by_cve(self): |
53 | | - """ |
54 | | - A single CVE can appear in multiple records across different operating systems, distributions, or package versions. This method groups all records with the same CVE together and skips entries that are invalid or marked as not affected. The result is a dictionary keyed by CVE ID, with each value containing the related records. |
55 | | - """ |
56 | | - self.cve_to_records = {} |
57 | | - skipped_invalid = 0 |
58 | | - skipped_non_affected = 0 |
59 | | - |
60 | | - for record in self.response: |
61 | | - cve_id = record.get("cve", "").strip() |
62 | | - if not cve_id: |
63 | | - self.log(f"Skipping record with empty CVE ID") |
64 | | - skipped_invalid += 1 |
65 | | - continue |
66 | | - |
67 | | - os_name = record.get("os_name", "").strip() |
68 | | - project_name = record.get("project_name", "").strip() |
69 | | - version = record.get("version", "").strip() |
70 | | - status = record.get("status", "").strip() |
71 | | - |
72 | | - if not all([os_name, project_name, version, status]): |
73 | | - self.log(f"Skipping {cve_id}: missing required fields") |
74 | | - skipped_invalid += 1 |
75 | | - continue |
76 | | - |
77 | | - # Skip records with non-affected statuses |
78 | | - if status in NON_AFFECTED_STATUSES: |
79 | | - skipped_non_affected += 1 |
80 | | - continue |
81 | | - |
82 | | - if status not in AFFECTED_STATUSES and status not in FIXED_STATUSES: |
83 | | - self.log(f"Skipping {cve_id}: unrecognized status '{status}'") |
84 | | - skipped_invalid += 1 |
85 | | - continue |
86 | | - |
87 | | - if cve_id not in self.cve_to_records: |
88 | | - self.cve_to_records[cve_id] = [] |
89 | | - self.cve_to_records[cve_id].append(record) |
90 | | - |
91 | | - total_skipped = skipped_invalid + skipped_non_affected |
92 | | - self.log( |
93 | | - f"Grouped {len(self.response):,d} records into {len(self.cve_to_records):,d} unique CVEs " |
94 | | - f"(skipped {total_skipped:,d}: {skipped_invalid:,d} invalid, " |
95 | | - f"{skipped_non_affected:,d} non-affected)" |
96 | | - ) |
| 43 | + def fetch_zip(self): |
| 44 | + self.log(f"Fetching `{self.url}`") |
| 45 | + self.advisory_zip = requests.get(self.url).content |
97 | 46 |
|
98 | 47 | def advisories_count(self) -> int: |
99 | | - return len(self.cve_to_records) |
100 | | - |
101 | | - def _create_purl(self, project_name: str, os_name: str) -> PackageURL: |
102 | | - normalized_os = os_name.lower().replace(" ", "-") |
103 | | - os_lower = os_name.lower() |
104 | | - |
105 | | - os_mapping = { |
106 | | - "ubuntu": ("deb", "ubuntu"), |
107 | | - "debian": ("deb", "debian"), |
108 | | - "centos": ("rpm", "centos"), |
109 | | - "almalinux": ("rpm", "almalinux"), |
110 | | - "rhel": ("rpm", "rhel"), |
111 | | - "oracle": ("rpm", "oracle"), |
112 | | - "cloudlinux": ("rpm", "cloudlinux"), |
113 | | - "alpine": ("apk", "alpine"), |
114 | | - "unknown": ("generic", "tuxcare"), |
115 | | - "tuxcare": ("generic", "tuxcare"), |
116 | | - } |
117 | | - |
118 | | - for keyword, (ptype, pns) in os_mapping.items(): |
119 | | - if keyword in os_lower: |
120 | | - pkg_type = ptype |
121 | | - namespace = pns |
122 | | - break |
123 | | - else: |
124 | | - return None |
125 | | - |
126 | | - qualifiers = {"distro": normalized_os} |
127 | | - |
128 | | - return PackageURL( |
129 | | - type=pkg_type, namespace=namespace, name=project_name, qualifiers=qualifiers |
130 | | - ) |
| 48 | + with ZipFile(BytesIO(self.advisory_zip)) as zip_file: |
| 49 | + advisory_count = sum( |
| 50 | + 1 |
| 51 | + for file in zip_file.namelist() |
| 52 | + if file.startswith("CLSA-") and file.endswith(".json") |
| 53 | + ) |
| 54 | + return advisory_count |
131 | 55 |
|
132 | 56 | def collect_advisories(self) -> Iterable[AdvisoryDataV2]: |
133 | | - for cve_id, records in self.cve_to_records.items(): |
134 | | - affected_packages = [] |
135 | | - severities = [] |
136 | | - date_published = None |
137 | | - all_records = [] |
138 | | - |
139 | | - for record in records: |
140 | | - os_name = record.get("os_name", "").strip() |
141 | | - project_name = record.get("project_name", "").strip() |
142 | | - version = record.get("version", "").strip() |
143 | | - score = record.get("score", "").strip() |
144 | | - severity = record.get("severity", "").strip() |
145 | | - status = record.get("status", "").strip() |
146 | | - last_updated = record.get("last_updated", "").strip() |
147 | | - |
148 | | - purl = self._create_purl(project_name, os_name) |
149 | | - if not purl: |
| 57 | + """Yield AdvisoryDataV2 using a zipped data dump of OSV data""" |
| 58 | + supported_ecosystems = [ |
| 59 | + # els_os |
| 60 | + "rpm", |
| 61 | + "deb", |
| 62 | + # els_lang |
| 63 | + "maven", |
| 64 | + "npm", |
| 65 | + "pypi", |
| 66 | + "composer", |
| 67 | + ] |
| 68 | + |
| 69 | + with ZipFile(BytesIO(self.advisory_zip)) as zip_file: |
| 70 | + for file_name in zip_file.namelist(): |
| 71 | + if not (file_name.startswith("CLSA-") and file_name.endswith(".json")): |
150 | 72 | self.log( |
151 | | - f"Skipping package {project_name} on {os_name} for {cve_id} - unexpected OS type" |
| 73 | + f"Unsupported TuxCare advisory data file: {file_name}", |
| 74 | + level=logging.ERROR, |
152 | 75 | ) |
153 | 76 | continue |
154 | | - |
155 | | - version_range_class = RANGE_CLASS_BY_SCHEMES.get(purl.type) |
156 | | - |
157 | | - try: |
158 | | - version_range = version_range_class.from_versions([version]) |
159 | | - except ValueError as e: |
160 | | - self.log(f"Failed to parse version {version} for {cve_id}: {e}") |
161 | | - continue |
162 | | - |
163 | | - affected_version_range = None |
164 | | - fixed_version_range = None |
165 | | - |
166 | | - if status in AFFECTED_STATUSES: |
167 | | - affected_version_range = version_range |
168 | | - elif status in FIXED_STATUSES: |
169 | | - fixed_version_range = version_range |
170 | | - |
171 | | - affected_packages.append( |
172 | | - AffectedPackageV2( |
173 | | - package=purl, |
174 | | - affected_version_range=affected_version_range, |
175 | | - fixed_version_range=fixed_version_range, |
| 77 | + with zip_file.open(file_name) as f: |
| 78 | + advisory_text = f.read() |
| 79 | + vul_info = json.loads(advisory_text) |
| 80 | + yield parse_advisory_data_v3( |
| 81 | + raw_data=vul_info, |
| 82 | + supported_ecosystems=supported_ecosystems, |
| 83 | + advisory_url=self.url, |
| 84 | + advisory_text=advisory_text.decode("utf-8"), |
176 | 85 | ) |
177 | | - ) |
178 | | - |
179 | | - # Severity is per-CVE hence we add it only once |
180 | | - if severity and score and not severities: |
181 | | - severities.append( |
182 | | - VulnerabilitySeverity( |
183 | | - system=GENERIC, |
184 | | - value=score, |
185 | | - scoring_elements=severity, |
186 | | - ) |
187 | | - ) |
188 | | - |
189 | | - if last_updated: |
190 | | - try: |
191 | | - current_date = parse(last_updated).replace(tzinfo=UTC) |
192 | | - if date_published is None or current_date > date_published: |
193 | | - date_published = current_date |
194 | | - except ValueError as e: |
195 | | - self.log(f"Failed to parse date {last_updated} for {cve_id}: {e}") |
196 | | - |
197 | | - all_records.append(record) |
198 | | - |
199 | | - if not affected_packages: |
200 | | - self.log(f"Skipping {cve_id} - no valid affected packages") |
201 | | - continue |
202 | | - |
203 | | - yield AdvisoryDataV2( |
204 | | - advisory_id=cve_id, |
205 | | - affected_packages=affected_packages, |
206 | | - severities=severities, |
207 | | - date_published=date_published, |
208 | | - url=f"https://cve.tuxcare.com/els/cve/{cve_id}", |
209 | | - original_advisory_text=json.dumps(all_records, indent=2, ensure_ascii=False), |
210 | | - ) |
0 commit comments