|
28 | 28 |
|
29 | 29 | import requests |
30 | 30 | from bs4 import BeautifulSoup |
| 31 | +from packageurl import PackageURL |
| 32 | +from univers.versions import AlpineLinuxVersion |
31 | 33 |
|
32 | 34 | from vulnerabilities.helpers import is_cve |
33 | 35 | from vulnerabilities.importer import AdvisoryData |
| 36 | +from vulnerabilities.importer import AffectedPackage |
34 | 37 | from vulnerabilities.importer import Importer |
35 | 38 | from vulnerabilities.importer import Reference |
36 | 39 |
|
37 | 40 |
|
38 | 41 | class AlpineImporter(Importer): |
| 42 | + spdx_license_expression = "CC-BY-SA-4.0" |
| 43 | + |
39 | 44 | def advisory_data(self) -> Iterable[AdvisoryData]: |
| 45 | + advisories = [] |
40 | 46 | advisory_links = fetch_advisory_links() |
41 | 47 | for link in advisory_links: |
42 | | - yield (process_link(link)) |
| 48 | + advisories.extend(process_link(link)) |
43 | 49 |
|
44 | 50 |
|
45 | 51 | def fetch_advisory_links(): |
@@ -74,26 +80,39 @@ def process_link(link) -> List[AdvisoryData]: |
74 | 80 | return advisories |
75 | 81 |
|
76 | 82 | for package in record["packages"]: |
| 83 | + assert isinstance(record["archs"], Iterable[str]) |
| 84 | + assert isinstance(record["reponame"], str) |
| 85 | + assert isinstance(record["distroversion"], str) |
77 | 86 | advisories.extend( |
78 | 87 | load_advisories( |
79 | 88 | package["pkg"], |
| 89 | + record["distroversion"], |
| 90 | + record["reponame"], |
| 91 | + record["archs"], |
80 | 92 | ) |
81 | 93 | ) |
82 | 94 | return advisories |
83 | 95 |
|
84 | 96 |
|
85 | 97 | def load_advisories( |
86 | 98 | pkg_infos: Mapping[str, Any], |
| 99 | + distroversion: str, |
| 100 | + reponame: str, |
| 101 | + archs: Iterable[str], |
87 | 102 | ) -> Iterable[AdvisoryData]: |
88 | 103 |
|
| 104 | + if not (pkg_infos.get("name") and isinstance(pkg_infos["name"], str)): |
| 105 | + return [] |
| 106 | + |
89 | 107 | assert isinstance(pkg_infos["secfixes"], Mapping[str, List]) |
90 | 108 |
|
91 | | - for fixed_vulns in pkg_infos["secfixes"].values(): |
| 109 | + for version, fixed_vulns in pkg_infos["secfixes"].items(): |
92 | 110 |
|
93 | 111 | if fixed_vulns is None: |
94 | 112 | continue |
95 | 113 |
|
96 | 114 | for vuln_ids in fixed_vulns: |
| 115 | + assert isinstance(vuln_ids, str) |
97 | 116 | vuln_ids = vuln_ids.split() |
98 | 117 | references = [] |
99 | 118 | for reference_id in vuln_ids[1:]: |
@@ -127,6 +146,20 @@ def load_advisories( |
127 | 146 | yield AdvisoryData( |
128 | 147 | summary="", |
129 | 148 | references=references, |
130 | | - affected_packages=[], |
| 149 | + affected_packages=[ |
| 150 | + AffectedPackage( |
| 151 | + package=PackageURL( |
| 152 | + type="alpine", |
| 153 | + name=pkg_infos["name"], |
| 154 | + qualifiers={ |
| 155 | + "arch": arch, |
| 156 | + "distroversion": distroversion, |
| 157 | + "reponame": reponame, |
| 158 | + }, |
| 159 | + ), |
| 160 | + fixed_version=AlpineLinuxVersion(version), |
| 161 | + ) |
| 162 | + for arch in archs |
| 163 | + ], |
131 | 164 | aliases=[vuln_ids[0] if is_cve(vuln_ids[0]) else ""], |
132 | 165 | ) |
0 commit comments