diff --git a/vulnerabilities/scraper/alpine_linux.py b/vulnerabilities/scraper/alpine_linux.py index 01713d6f0..4331b0f3d 100644 --- a/vulnerabilities/scraper/alpine_linux.py +++ b/vulnerabilities/scraper/alpine_linux.py @@ -1,42 +1,75 @@ import itertools import saneyaml +from schema import Regex, Or, Schema from urllib.request import urlopen from io import BytesIO from zipfile import ZipFile -ALPINE_DB_URL = 'https://gitlab.alpinelinux.org/alpine/infra/alpine-secdb/-/\ -archive/master/alpine-secdb-master.zip' +ALPINE_DB_URL = "https://gitlab.alpinelinux.org/alpine/infra/alpine-secdb/-/\ +archive/master/alpine-secdb-master.zip" def alpine_advisories(url): with urlopen(url) as response: with ZipFile(BytesIO(response.read())) as zf: for path in zf.namelist(): - if path.endswith('main.yaml'): + if path.endswith("main.yaml"): yield saneyaml.load(zf.open(path)) +def validate_schema(advisory_dict): + scheme = { + "distroversion": Regex(r"v\d.\d*"), + "reponame": "main", + "archs": list, + "packages": [ + { + "pkg": { + "name": str, + "secfixes": { + str: Or( + [ + Or( + Regex(r"CVE.\d+-\d+"), + Regex(r"XSA-\d{3}"), + Regex(r"ZBX-\d{4}"), + Regex(r"wnpa-sec-\d{4}-\d{2}"), + ) + ], + "", + ), + }, + } + } + ], + object: object, + } + Schema(scheme).validate(advisory_dict) + + def import_vulnerabilities(): vulnerability_package_dicts = [] for vulnerability in alpine_advisories(ALPINE_DB_URL): - for pkg_details in vulnerability['packages']: - package_name = pkg_details['pkg']['name'] - for version, fixed_cves in pkg_details['pkg']['secfixes'].items(): + validate_schema(vulnerability) + for pkg_details in vulnerability["packages"]: + package_name = pkg_details["pkg"]["name"] + for version, fixed_cves in pkg_details["pkg"]["secfixes"].items(): # ['CVE-2016-9932 XSA-200', 'CVE-2016-9815','CVE-????-?????'] after mapping # the split function to above list all_cves = list(map(lambda x: x.split(), fixed_cves)) # it becomes [['CVE-2016-9932','XSA-200'], ['CVE-2016-9815'],['CVE-????-?????']] for index, vuln_grp in enumerate(all_cves): all_cves[index] = list( - filter(lambda x: 'CVE-????-?????' not in x, vuln_grp)) + filter(lambda x: "CVE-????-?????" not in x, vuln_grp) + ) all_cves = [i for i in all_cves if i and len(i) <= 2] # this data consists lots of 'CVE-????-?????' to denote vulnerabilities # with unassigned CVE ids , we filter out these as well as other garbage data vulnerability_package_dicts.append( { - 'package_name': package_name, - 'vuln_ids': all_cves, - 'fixed_version': version, + "package_name": package_name, + "vuln_ids": all_cves, + "fixed_version": version, } ) return vulnerability_package_dicts