diff --git a/vulnerabilities/data_source.py b/vulnerabilities/data_source.py index 0dbfd7677..091b72394 100644 --- a/vulnerabilities/data_source.py +++ b/vulnerabilities/data_source.py @@ -43,6 +43,17 @@ from vulnerabilities.oval_parser import OvalParser +@dataclasses.dataclass +class Reference: + + url: str = '' + reference_id: str = '' + + def __post_init__(self): + if not any([self.url, self.reference_id]): + raise TypeError + + @dataclasses.dataclass class Advisory: """ @@ -57,16 +68,14 @@ class Advisory: summary: str impacted_package_urls: Iterable[PackageURL] resolved_package_urls: Iterable[PackageURL] = dataclasses.field(default_factory=list) - reference_urls: Sequence[str] = dataclasses.field(default_factory=list) - reference_ids: Sequence[str] = dataclasses.field(default_factory=list) + vuln_references: List[Reference] = dataclasses.field(default_factory=list) cve_id: Optional[str] = None def __hash__(self): - s = '{}{}{}{}{}'.format( + s = '{}{}{}{}'.format( self.summary, ''.join(sorted([str(p) for p in self.impacted_package_urls])), ''.join(sorted([str(p) for p in self.resolved_package_urls])), - ''.join(sorted(self.reference_urls)), self.cve_id, ) return hash(s) @@ -472,7 +481,8 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis description = definition_data['description'] affected_purls = set() safe_purls = set() - urls = definition_data['reference_urls'] + references = [Reference(url=url) + for url in definition_data['reference_urls']] for test_data in definition_data['test_data']: for package in test_data['package_list']: @@ -511,5 +521,5 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis impacted_package_urls=affected_purls, resolved_package_urls=safe_purls, cve_id=vuln_id, - reference_urls=urls)) + vuln_references=references)) return all_adv diff --git a/vulnerabilities/import_runner.py b/vulnerabilities/import_runner.py index c5331d4e0..c58cdeee4 100644 --- a/vulnerabilities/import_runner.py +++ b/vulnerabilities/import_runner.py @@ -1,5 +1,5 @@ # -# Copyright (c) 2017 nexB Inc. and others. All rights reserved. +# Copyright (c) nexB Inc. and others. All rights reserved. # http://nexb.com and https://github.com/nexB/vulnerablecode/ # The VulnerableCode software is licensed under the Apache License version 2.0. # Data generated with VulnerableCode require an acknowledgment. @@ -18,7 +18,7 @@ # OR CONDITIONS OF ANY KIND, either express or implied. No content created from # VulnerableCode should be considered or used as legal advice. Consult an Attorney # for any legal advice. -# VulnerableCode is a free software code scanning tool from nexB Inc. and others. +# VulnerableCode is a free software tool from nexB Inc. and others. # Visit https://github.com/nexB/vulnerablecode/ for support and download. import dataclasses @@ -51,7 +51,7 @@ class VulnerabilityReferenceInserter: vulnerability: models.Vulnerability reference_id: Optional[str] = '' - url: Optional[str] = '' + url: Optional[str] = '' def __post_init__(self): if not any([self.reference_id, self.url]): @@ -165,37 +165,60 @@ def _process_added_advisories(data_source: DataSource) -> None: def _create_vulnerability_and_references(advisory: Advisory): vuln, vuln_created = _get_or_create_vulnerability(advisory) - vuln_references = set() + new_vuln_references = set() if vuln_created: # This means vulnerability didn't previously exist in the DB, so add # the references to bulk create queue without any hesitation - for id_ in advisory.reference_ids: - vuln_references.add(VulnerabilityReferenceInserter(vulnerability=vuln, reference_id=id_)) # nopep8 - - for url in advisory.reference_urls: - vuln_references.add(VulnerabilityReferenceInserter(vulnerability=vuln, url=url)) + for ref in advisory.vuln_references: + new_vuln_references.add( + VulnerabilityReferenceInserter( + vulnerability=vuln, + **dataclasses.asdict(ref))) else: - vuln_refs_qs = models.VulnerabilityReference.objects.filter(vulnerability=vuln) - vuln_ids = {ref.reference_id for ref in vuln_refs_qs} - vuln_urls = {ref.url for ref in vuln_refs_qs} + # build indexes of existing references to either update or create new ones + # and avoid creating duplicates + + existing_vuln_references = models.VulnerabilityReference.objects.filter(vulnerability=vuln) + refs_by_id_url = { + (ref.id, ref.url): ref for ref in existing_vuln_references if ref.id and ref.url} + refs_by_id = {ref.id: ref for ref in existing_vuln_references if ref.id} + refs_by_url = {ref.url: ref for ref in existing_vuln_references if ref.url} + + for adv_ref in advisory.vuln_references: + if adv_ref.reference_id and adv_ref.url: + existing_ref = refs_by_id_url.get((adv_ref.reference_id, adv_ref.url)) + + elif adv_ref.reference_id: + existing_ref = refs_by_id.get(adv_ref.reference_id) + + elif adv_ref.url: + existing_ref = refs_by_url.get(adv_ref.url) - for id_ in advisory.reference_ids: - # Add the item preventing duplicates pass to through. - if id_ not in vuln_ids: - vuln_ids.add(id_) - vuln_references.add(VulnerabilityReferenceInserter( - vulnerability=vuln, reference_id=id_)) + if existing_ref: + # update ref as needed + if not adv_ref.reference_id or not adv_ref.url: + continue - for url in advisory.reference_urls: - # Add the item preventing duplicates pass to through. - if url not in vuln_urls: - vuln_urls.add(url) - vuln_references.add(VulnerabilityReferenceInserter( - vulnerability=vuln, url=url)) + if adv_ref.reference_id and not existing_ref.reference_id: + existing_ref.reference_id = adv_ref.reference_id - return vuln, vuln_created, vuln_references + if adv_ref.url and not existing_ref.url: + existing_ref.url = adv_ref.url + + existing_ref.save() + + else: + new_vuln_references.add( + VulnerabilityReferenceInserter( + vulnerability=vuln, + reference_id=adv_ref.reference_id, + url=adv_ref.url, + ) + ) + + return vuln, vuln_created, new_vuln_references def _create_pkg_vuln_refs(vuln: models.Vulnerability, vuln_created: bool, purls: Sequence[PackageURL], is_vulnerable: bool): # nopep8 @@ -203,7 +226,7 @@ def _create_pkg_vuln_refs(vuln: models.Vulnerability, vuln_created: bool, purls: for purl in purls: pkg, pkg_created = _get_or_create_package(purl) vuln_pkg_ref = PackageRelatedVulnerabilityInserter( - package=pkg, vulnerability=vuln, is_vulnerable=is_vulnerable) + package=pkg, vulnerability=vuln, is_vulnerable=is_vulnerable) if pkg_created or vuln_created: new_refs.add(vuln_pkg_ref) @@ -371,13 +394,12 @@ def _insert_vulnerabilities_and_references(batch: Set[Advisory]) -> Set[models.V vulnerabilities.add(vuln) - for id_ in advisory.reference_ids: - models.VulnerabilityReference.objects.get_or_create( - vulnerability=vuln, reference_id=id_) + # TODO: Refactor this method to be atleast as efficient as + # one used by _process_updated_advisories - for url in advisory.reference_urls: + for ref in advisory.vuln_references: models.VulnerabilityReference.objects.get_or_create( - vulnerability=vuln, url=url) + vulnerability=vuln, reference_id=ref.reference_id, url=ref.url) return vulnerabilities diff --git a/vulnerabilities/importers/alpine_linux.py b/vulnerabilities/importers/alpine_linux.py index 36b95359b..78085db26 100644 --- a/vulnerabilities/importers/alpine_linux.py +++ b/vulnerabilities/importers/alpine_linux.py @@ -34,6 +34,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import GitDataSource +from vulnerabilities.data_source import Reference def validate_schema(advisory_dict): @@ -53,14 +54,12 @@ def validate_schema(advisory_dict): Regex(r"XSA-\d{3}"), Regex(r"ZBX-\d{4}"), Regex(r"wnpa-sec-\d{4}-\d{2}"), - - ) ], "", # FIXME: Remove the None when below issue gets fixed # https://gitlab.alpinelinux.org/alpine/infra/alpine-secdb/-/issues/1 - None + None, ), }, } @@ -72,13 +71,13 @@ def validate_schema(advisory_dict): class AlpineDataSource(GitDataSource): - def __enter__(self): super(AlpineDataSource, self).__enter__() - if not getattr(self, '_added_files', None): + if not getattr(self, "_added_files", None): self._added_files, self._updated_files = self.file_changes( - recursive=True, file_ext='yaml') + recursive=True, file_ext="yaml" + ) def updated_advisories(self) -> Set[Advisory]: files = self._updated_files.union(self._added_files) @@ -94,55 +93,80 @@ def _process_file(self, path) -> List[Advisory]: with open(path) as f: record = yaml.safe_load(f) - if record['packages'] is None: + if record["packages"] is None: return advisories validate_schema(record) - for p in record['packages']: - advisories.extend(self._load_advisories( - p['pkg'], - record['distroversion'], - record['reponame'], - record['archs'], - )) + for p in record["packages"]: + advisories.extend( + self._load_advisories( + p["pkg"], record["distroversion"], record["reponame"], record["archs"], + ) + ) return advisories def _load_advisories( - self, - pkg_infos: Mapping[str, Any], - distroversion: str, - reponame: str, - archs: Iterable[str], + self, pkg_infos: Mapping[str, Any], distroversion: str, reponame: str, archs: Iterable[str], ) -> List[Advisory]: advisories = [] - for version, fixed_vulns in pkg_infos['secfixes'].items(): + for version, fixed_vulns in pkg_infos["secfixes"].items(): if fixed_vulns is None: continue resolved_purls = { PackageURL( - name=pkg_infos['name'], - type='alpine', + name=pkg_infos["name"], + type="alpine", version=version, - qualifiers={ - 'arch': arch, 'distroversion': distroversion, 'reponame': reponame}, + qualifiers={"arch": arch, "distroversion": distroversion, "reponame": reponame}, ) for arch in archs } for vuln_ids in fixed_vulns: vuln_ids = vuln_ids.split() - - advisories.append(Advisory( - summary='', - impacted_package_urls=[], - resolved_package_urls=resolved_purls, - reference_ids=vuln_ids[1:], - cve_id=vuln_ids[0] if vuln_ids[0] != 'CVE-????-?????' else None, - )) + references = [] + for reference_id in vuln_ids[1:]: + + if reference_id.startswith("XSA"): + xsa_id = reference_id.split("-")[-1] + references.append( + Reference( + reference_id=reference_id, + url="https://xenbits.xen.org/xsa/advisory-{}.html".format(xsa_id), + ) + ) + + elif reference_id.startswith("ZBX"): + references.append( + Reference( + reference_id=reference_id, + url="https://support.zabbix.com/browse/{}".format(reference_id), + ) + ) + + elif reference_id.startswith("wnpa-sec"): + references.append( + Reference( + reference_id=reference_id, + url="https://www.wireshark.org/security/{}.html".format( + reference_id + ), + ) + ) + + advisories.append( + Advisory( + summary="", + impacted_package_urls=[], + resolved_package_urls=resolved_purls, + vuln_references=references, + cve_id=vuln_ids[0] if vuln_ids[0] != "CVE-????-?????" else None, + ) + ) return advisories diff --git a/vulnerabilities/importers/archlinux.py b/vulnerabilities/importers/archlinux.py index 37c80b658..95e674735 100644 --- a/vulnerabilities/importers/archlinux.py +++ b/vulnerabilities/importers/archlinux.py @@ -1,5 +1,5 @@ # -# Copyright (c) 2017 nexB Inc. and others. All rights reserved. +# Copyright (c) nexB Inc. and others. All rights reserved. # http://nexb.com and https://github.com/nexB/vulnerablecode/ # The VulnerableCode software is licensed under the Apache License version 2.0. # Data generated with VulnerableCode require an acknowledgment. @@ -18,7 +18,7 @@ # OR CONDITIONS OF ANY KIND, either express or implied. No content created from # VulnerableCode should be considered or used as legal advice. Consult an Attorney # for any legal advice. -# VulnerableCode is a free software code scanning tool from nexB Inc. and others. +# VulnerableCode is a free software tool from nexB Inc. and others. # Visit https://github.com/nexB/vulnerablecode/ for support and download. import dataclasses import json @@ -31,23 +31,24 @@ from packageurl import PackageURL from schema import Regex, Schema, Or -from vulnerabilities.data_source import DataSource, DataSourceConfiguration, Advisory +from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import DataSource +from vulnerabilities.data_source import DataSourceConfiguration +from vulnerabilities.data_source import Reference def validate_schema(advisory_dict): scheme = { - - 'advisories': list, - 'affected': str, - 'fixed': Or(None, str), - 'issues': [Regex(r'CVE-\d+-\d+')], - 'name': str, - 'packages': [str], - 'status': str, - 'ticket': object, - 'type': str, - 'severity': str, - + "advisories": list, + "affected": str, + "fixed": Or(None, str), + "issues": [Regex(r"CVE-\d+-\d+")], + "name": str, + "packages": [str], + "status": str, + "ticket": object, + "type": str, + "severity": str, } Schema(scheme).validate(advisory_dict) @@ -83,32 +84,49 @@ def _fetch(self) -> Iterable[Mapping]: def _parse(self, record) -> List[Advisory]: advisories = [] - for cve_id in record['issues']: + for cve_id in record["issues"]: impacted_purls, resolved_purls = set(), set() - for name in record['packages']: - impacted_purls.add(PackageURL( - name=name, - type='pacman', - namespace='archlinux', - version=record['affected'], - )) - - if record['fixed']: - resolved_purls.add(PackageURL( - name=name, - type='pacman', - namespace='archlinux', - version=record['fixed'], - )) - - reference_urls = [f'https://security.archlinux.org/{a}' for a in record['advisories']] - - advisories.append(Advisory( - cve_id=cve_id, - summary='', - impacted_package_urls=impacted_purls, - resolved_package_urls=resolved_purls, - reference_urls=reference_urls, - )) + for name in record["packages"]: + impacted_purls.add( + PackageURL( + name=name, type="pacman", namespace="archlinux", version=record["affected"], + ) + ) + + if record["fixed"]: + resolved_purls.add( + PackageURL( + name=name, + type="pacman", + namespace="archlinux", + version=record["fixed"], + ) + ) + + vuln_references = [] + vuln_references.append( + Reference( + reference_id=record["name"], + url="https://security.archlinux.org/{}".format(record["name"]), + ) + ) + + for ref in record["advisories"]: + vuln_references.append( + Reference( + reference_id=ref, + url="https://security.archlinux.org/{}".format(ref), + ) + ) + + advisories.append( + Advisory( + cve_id=cve_id, + summary="", + impacted_package_urls=impacted_purls, + resolved_package_urls=resolved_purls, + vuln_references=vuln_references, + ) + ) return advisories diff --git a/vulnerabilities/importers/debian.py b/vulnerabilities/importers/debian.py index b9cd3d5ec..0715d1c8b 100644 --- a/vulnerabilities/importers/debian.py +++ b/vulnerabilities/importers/debian.py @@ -37,6 +37,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import DataSource from vulnerabilities.data_source import DataSourceConfiguration +from vulnerabilities.data_source import Reference def validate_schema(advisory_dict): @@ -131,18 +132,21 @@ def _parse(self, pkg_name: str, records: Mapping[str, Any]) -> List[Advisory]: qualifiers={'distro': release_name}, )) - reference_urls = [] + references = [] debianbug = record.get('debianbug') if debianbug: bug_url = f'https://bugs.debian.org/cgi-bin/bugreport.cgi?bug={debianbug}' - reference_urls.append(bug_url) + references.append(Reference( + url=bug_url, + reference_id=debianbug + )) advisories.append(Advisory( cve_id=cve_id, summary=record.get('description', ''), impacted_package_urls=impacted_purls, resolved_package_urls=resolved_purls, - reference_urls=reference_urls, + vuln_references=references, )) return advisories diff --git a/vulnerabilities/importers/gentoo.py b/vulnerabilities/importers/gentoo.py index 95255d34e..57b273b9c 100644 --- a/vulnerabilities/importers/gentoo.py +++ b/vulnerabilities/importers/gentoo.py @@ -28,6 +28,7 @@ from vulnerabilities.data_source import GitDataSource from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference class GentooDataSource(GitDataSource): @@ -51,6 +52,13 @@ def process_file(self, file): xml_data = {} xml_root = ET.parse(file).getroot() glsa = "GLSA-" + xml_root.attrib["id"] + vuln_reference = [ + Reference( + reference_id=glsa, + url="https://security.gentoo.org/glsa/{}".format(xml_root.attrib["id"]), + ) + ] + for child in xml_root: if child.tag == "references": xml_data["cves"] = self.cves_from_reference(child) @@ -73,7 +81,7 @@ def process_file(self, file): summary=xml_data["description"], impacted_package_urls=xml_data["affected_purls"], resolved_package_urls=xml_data["unaffected_purls"], - reference_ids=[glsa], + vuln_references=vuln_reference, ) advisory_list.append(advisory) return advisory_list @@ -96,11 +104,8 @@ def affected_and_safe_purls(affected_elem): for pkg in affected_elem: for info in pkg: - pkg_ns, pkg_name, = pkg.attrib["name"].split('/') - purl = PackageURL( - type="ebuild", name=pkg_name, version=info.text, - namespace=pkg_ns - ) + pkg_ns, pkg_name, = pkg.attrib["name"].split("/") + purl = PackageURL(type="ebuild", name=pkg_name, version=info.text, namespace=pkg_ns) if info.attrib.get("range"): if len(info.attrib.get("range")) > 2: diff --git a/vulnerabilities/importers/github.py b/vulnerabilities/importers/github.py index 92e0063b9..66fca5d0d 100644 --- a/vulnerabilities/importers/github.py +++ b/vulnerabilities/importers/github.py @@ -38,6 +38,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import DataSource from vulnerabilities.data_source import DataSourceConfiguration +from vulnerabilities.data_source import Reference # set of all possible values of first '%s' = {'MAVEN','COMPOSER', 'NUGET'} @@ -167,24 +168,33 @@ def process_response(self) -> List[Advisory]: ) affected_purls = { - PackageURL(name=pkg_name, namespace=ns, version=version, type=pkg_type) + PackageURL(name=pkg_name, namespace=ns, + version=version, type=pkg_type) for version in aff_vers } unaffected_purls = { - PackageURL(name=pkg_name, namespace=ns, version=version, type=pkg_type) + PackageURL(name=pkg_name, namespace=ns, + version=version, type=pkg_type) for version in unaff_vers } cve_ids = set() - ref_ids = set() + vuln_references = [] vuln_desc = adv["node"]["advisory"]["summary"] for vuln in adv["node"]["advisory"]["identifiers"]: if vuln["type"] == "CVE": cve_ids.add(vuln["value"]) - else: - ref_ids.add(vuln["value"]) + + elif vuln["type"] == "GHSA": + ghsa = vuln['value'] + vuln_references.append(Reference( + reference_id=ghsa, + url="https://github.com/advisories/{}".format( + ghsa) + )) + for cve_id in cve_ids: adv_list.append( Advisory( @@ -192,7 +202,7 @@ def process_response(self) -> List[Advisory]: summary=vuln_desc, impacted_package_urls=affected_purls, resolved_package_urls=unaffected_purls, - reference_ids=ref_ids, + vuln_references=vuln_references, ) ) return adv_list @@ -200,7 +210,8 @@ def process_response(self) -> List[Advisory]: @staticmethod def categorize_versions(version_range: str, all_versions: Set[str]) -> Tuple[Set[str], Set[str]]: # nopep8 version_range = RangeSpecifier(version_range) - affected_versions = {version for version in all_versions if version in version_range} + affected_versions = { + version for version in all_versions if version in version_range} return (affected_versions, all_versions - affected_versions) diff --git a/vulnerabilities/importers/npm.py b/vulnerabilities/importers/npm.py index 2862ba64b..86a1e063b 100644 --- a/vulnerabilities/importers/npm.py +++ b/vulnerabilities/importers/npm.py @@ -37,6 +37,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import GitDataSource +from vulnerabilities.data_source import Reference NPM_URL = 'https://registry.npmjs.org{}' @@ -67,41 +68,44 @@ def process_file(self, file) -> List[Advisory]: with open(file) as f: record = json.load(f) advisories = [] - package_name = record['module_name'] + package_name = record["module_name"] all_versions = self.versions.get(package_name) - aff_range = record.get('vulnerable_versions', '') - fixed_range = record.get('patched_versions', '') + aff_range = record.get("vulnerable_versions", "") + fixed_range = record.get("patched_versions", "") impacted_versions, resolved_versions = categorize_versions( - all_versions, - aff_range, - fixed_range + all_versions, aff_range, fixed_range ) impacted_purls = _versions_to_purls(package_name, impacted_versions) resolved_purls = _versions_to_purls(package_name, resolved_versions) - - for cve_id in record.get('cves') or ['']: - advisories.append(Advisory( - summary=record.get('overview', ''), - cve_id=cve_id, - impacted_package_urls=impacted_purls, - resolved_package_urls=resolved_purls, - reference_urls=[NPM_URL.format(f'/-/npm/v1/advisories/{record["id"]}')], - )) - + vuln_reference = [ + Reference( + url=NPM_URL.format(f'/-/npm/v1/advisories/{record["id"]}'), + reference_id=record["id"], + ) + ] + + for cve_id in record.get("cves") or [""]: + advisories.append( + Advisory( + summary=record.get("overview", ""), + cve_id=cve_id, + impacted_package_urls=impacted_purls, + resolved_package_urls=resolved_purls, + vuln_references=vuln_reference, + ) + ) return advisories def _versions_to_purls(package_name, versions): - purls = {f'pkg:npm/{quote(package_name)}@{v}' for v in versions} + purls = {f"pkg:npm/{quote(package_name)}@{v}" for v in versions} return {PackageURL.from_string(s) for s in purls} def categorize_versions( - all_versions: Set[str], - aff_version_range: str, - fixed_version_range: str, + all_versions: Set[str], aff_version_range: str, fixed_version_range: str, ) -> Tuple[Set[str], Set[str]]: """ Seperate list of affected versions and unaffected versions from all versions @@ -141,9 +145,9 @@ def get(self, package_name: str) -> Set[str]: if package_name not in self.cache: releases = set() try: - with urlopen(f'https://registry.npmjs.org/{package_name}') as response: + with urlopen(f"https://registry.npmjs.org/{package_name}") as response: data = json.load(response) - releases = {v for v in data.get('versions', {})} + releases = {v for v in data.get("versions", {})} except HTTPError as e: if e.code == 404: # NPM registry has no data regarding this package, we skip these diff --git a/vulnerabilities/importers/openssl.py b/vulnerabilities/importers/openssl.py index 778b747fc..ffc01ec15 100644 --- a/vulnerabilities/importers/openssl.py +++ b/vulnerabilities/importers/openssl.py @@ -28,6 +28,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import DataSource +from vulnerabilities.data_source import Reference import xml.etree.ElementTree as ET @@ -49,46 +50,51 @@ def to_advisories(xml_response: str) -> Set[Advisory]: root = ET.fromstring(xml_response) for element in root: if element.tag == "issue": - cve_id = '' - summary = '' + cve_id = "" + summary = "" safe_pkg_versions = [] vuln_pkg_versions = [] ref_urls = [] for info in element: - if info.tag == 'cve': - cve_id = 'CVE-' + info.attrib.get('name') + if info.tag == "cve": + cve_id = "CVE-" + info.attrib.get("name") - if info.tag == 'affects': + if info.tag == "affects": # Vulnerable package versions - vuln_pkg_versions.append(info.attrib.get('version')) + vuln_pkg_versions.append(info.attrib.get("version")) - if info.tag == 'fixed': + if info.tag == "fixed": # Fixed package versions - safe_pkg_versions.append(info.attrib.get('version')) + safe_pkg_versions.append(info.attrib.get("version")) if info: - commit_hash = info[0].attrib['hash'] - ref_urls.append("https://github.com/openssl/openssl/commit/" - + commit_hash) - if info.tag == 'description': + commit_hash = info[0].attrib["hash"] + ref_urls.append( + Reference( + url="https://github.com/openssl/openssl/commit/" + commit_hash + ) + ) + if info.tag == "description": # Description - summary = re.sub(r'\s+', ' ', info.text).strip() + summary = re.sub(r"\s+", " ", info.text).strip() - safe_purls = {PackageURL(name=pkg_name, - type=pkg_type, - version=version) - for version in safe_pkg_versions} - vuln_purls = {PackageURL(name=pkg_name, - type=pkg_type, - version=version) - for version in vuln_pkg_versions} + safe_purls = { + PackageURL(name=pkg_name, type=pkg_type, version=version) + for version in safe_pkg_versions + } + vuln_purls = { + PackageURL(name=pkg_name, type=pkg_type, version=version) + for version in vuln_pkg_versions + } - advisory = Advisory(cve_id=cve_id, - summary=summary, - impacted_package_urls=vuln_purls, - resolved_package_urls=safe_purls, - reference_urls=ref_urls) + advisory = Advisory( + cve_id=cve_id, + summary=summary, + impacted_package_urls=vuln_purls, + resolved_package_urls=safe_purls, + vuln_references=ref_urls, + ) advisories.append(advisory) return advisories diff --git a/vulnerabilities/importers/redhat.py b/vulnerabilities/importers/redhat.py index 98910582f..d6e69b922 100644 --- a/vulnerabilities/importers/redhat.py +++ b/vulnerabilities/importers/redhat.py @@ -27,6 +27,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import DataSource from vulnerabilities.data_source import DataSourceConfiguration +from vulnerabilities.data_source import Reference class RedhatDataSource(DataSource): @@ -66,17 +67,35 @@ def fetch(): def to_advisory(advisory_data): affected_purls = [] - if advisory_data.get('affected_packages'): - for rpm in advisory_data['affected_packages']: + if advisory_data.get("affected_packages"): + for rpm in advisory_data["affected_packages"]: if rpm_to_purl(rpm): affected_purls.append(rpm_to_purl(rpm)) + references = [] + if advisory_data.get("bugzilla"): + bugzilla = advisory_data.get("bugzilla") + references.append( + Reference( + url="https://bugzilla.redhat.com/show_bug.cgi?id={}".format(bugzilla), + reference_id=bugzilla, + ) + ) + + for rhsa in advisory_data["advisories"]: + references.append( + Reference( + url="https://access.redhat.com/errata/{}".format(rhsa), reference_id=rhsa, + ) + ) + + references.append(Reference(url=advisory_data["resource_url"])) + return Advisory( - summary=advisory_data['bugzilla_description'], - cve_id=advisory_data['CVE'], - reference_ids=advisory_data['advisories'], + summary=advisory_data["bugzilla_description"], + cve_id=advisory_data["CVE"], impacted_package_urls=affected_purls, - reference_urls=[advisory_data['resource_url']], + vuln_references=references, ) @@ -84,11 +103,11 @@ def rpm_to_purl(rpm_string): # Red Hat uses `-:0` instead of just `-` to separate # package name and version - components = rpm_string.split('-0:') + components = rpm_string.split("-0:") if len(components) != 2: return name, version = components if version[0].isdigit(): - return PackageURL(name=name, type='rpm', version=version, namespace="redhat") + return PackageURL(name=name, type="rpm", version=version, namespace="redhat") diff --git a/vulnerabilities/importers/retiredotnet.py b/vulnerabilities/importers/retiredotnet.py index a01e1a0a5..76322e8c2 100644 --- a/vulnerabilities/importers/retiredotnet.py +++ b/vulnerabilities/importers/retiredotnet.py @@ -30,6 +30,7 @@ from vulnerabilities.data_source import GitDataSource from vulnerabilities.data_source import GitDataSourceConfiguration from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference class RetireDotnetDataSource(GitDataSource): @@ -90,9 +91,13 @@ def process_file(self, path) -> List[Advisory]: version=pkg['fix'], type='nuget')) + vuln_reference = [Reference( + url=json_doc['link'], + )] + return Advisory( summary=json_doc['description'], impacted_package_urls=affected_purls, resolved_package_urls=fixed_purls, - reference_urls=[json_doc['link']], - cve_id=vuln_id) + cve_id=vuln_id, + vuln_references=vuln_reference) diff --git a/vulnerabilities/importers/ruby.py b/vulnerabilities/importers/ruby.py index 86d01b9bc..3b36b2379 100644 --- a/vulnerabilities/importers/ruby.py +++ b/vulnerabilities/importers/ruby.py @@ -32,6 +32,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import GitDataSource +from vulnerabilities.data_source import Reference class RubyDataSource(GitDataSource): @@ -104,11 +105,17 @@ def process_file(self, path) -> List[Advisory]: version=version, ) for version in safe_versions} + references = [] + if record.get('url'): + references.append( + Reference(url=record.get('url')) + ) + return Advisory( summary=record.get('description', ''), impacted_package_urls=impacted_purls, resolved_package_urls=resolved_purls, - reference_urls=[record.get('url', '')], + vuln_references=references, cve_id=cve_id ) diff --git a/vulnerabilities/importers/rust.py b/vulnerabilities/importers/rust.py index dc5ac2e42..886efce4c 100644 --- a/vulnerabilities/importers/rust.py +++ b/vulnerabilities/importers/rust.py @@ -34,24 +34,24 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import GitDataSource +from vulnerabilities.data_source import Reference class RustDataSource(GitDataSource): - def __enter__(self): super(RustDataSource, self).__enter__() - if not getattr(self, '_added_files', None): + if not getattr(self, "_added_files", None): self._added_files, self._updated_files = self.file_changes( - subdir='crates', # TODO Consider importing the advisories for cargo, etc as well. + subdir="crates", # TODO Consider importing the advisories for cargo, etc as well. recursive=True, - file_ext='toml', + file_ext="toml", ) @property def crates_api(self): - if not hasattr(self, '_crates_api'): - setattr(self, '_crates_api', VersionAPI()) + if not hasattr(self, "_crates_api"): + setattr(self, "_crates_api", VersionAPI()) return self._crates_api def added_advisories(self) -> Set[Advisory]: @@ -61,10 +61,10 @@ def updated_advisories(self) -> Set[Advisory]: return self._load_advisories(self._updated_files) def _load_advisories(self, files) -> Set[Advisory]: - files = [f for f in files if not f.endswith('-0000.toml')] # skip temporary files + files = [f for f in files if not f.endswith("-0000.toml")] # skip temporary files while files: - batch, files = files[:self.batch_size], files[self.batch_size:] + batch, files = files[: self.batch_size], files[self.batch_size:] advisories = set() @@ -77,49 +77,60 @@ def _load_advisories(self, files) -> Set[Advisory]: def _load_advisory(self, path: str) -> Optional[Advisory]: with open(path) as f: record = toml.load(f) - advisory = record.get('advisory', {}) + advisory = record.get("advisory", {}) + + references = [] + crate_name = advisory["package"] + if advisory.get("url"): + references.append(Reference(url=advisory["url"])) - crate_name = advisory['package'] - reference_url = advisory.get('url', '') all_versions = self.crates_api.get(crate_name) - affected_ranges = {RangeSpecifier(r) for r - in chain.from_iterable( - record.get('affected', {}).get('functions', {}).values())} + affected_ranges = { + RangeSpecifier(r) + for r in chain.from_iterable(record.get("affected", {}).get("functions", {}).values()) + } - unaffected_ranges = {RangeSpecifier(r) for r - in record.get('versions', {}).get('unaffected', [])} - resolved_ranges = {RangeSpecifier(r) for r - in record.get('versions', {}).get('patched', [])} + unaffected_ranges = { + RangeSpecifier(r) for r in record.get("versions", {}).get("unaffected", []) + } + resolved_ranges = {RangeSpecifier(r) for r in record.get("versions", {}).get("patched", [])} unaffected, affected = categorize_versions( - all_versions, unaffected_ranges, affected_ranges, resolved_ranges) + all_versions, unaffected_ranges, affected_ranges, resolved_ranges + ) - impacted_purls = {PackageURL(type='cargo', name=crate_name, version=v) for v in affected} - resolved_purls = {PackageURL(type='cargo', name=crate_name, version=v) for v in unaffected} + impacted_purls = {PackageURL(type="cargo", name=crate_name, version=v) for v in affected} + resolved_purls = {PackageURL(type="cargo", name=crate_name, version=v) for v in unaffected} cve_id = None - if 'aliases' in advisory: - for alias in advisory['aliases']: - if alias.startswith('CVE-'): + if "aliases" in advisory: + for alias in advisory["aliases"]: + if alias.startswith("CVE-"): cve_id = alias break + references.append( + Reference( + reference_id=advisory["id"], + url="https://rustsec.org/advisories/{}.html".format(advisory["id"]), + ) + ) + return Advisory( - summary=advisory.get('description', ''), + summary=advisory.get("description", ""), impacted_package_urls=impacted_purls, resolved_package_urls=resolved_purls, - reference_urls=[reference_url] if reference_url else [], - reference_ids=[advisory['id']], cve_id=cve_id, + vuln_references=references, ) def categorize_versions( - all_versions: Set[str], - unaffected_versions: Set[RangeSpecifier], - affected_versions: Set[RangeSpecifier], - resolved_versions: Set[RangeSpecifier], + all_versions: Set[str], + unaffected_versions: Set[RangeSpecifier], + affected_versions: Set[RangeSpecifier], + resolved_versions: Set[RangeSpecifier], ) -> Tuple[Set[str], Set[str]]: """ Categorize all versions of a crate according to the given version ranges. @@ -162,10 +173,10 @@ def get(self, package_name: str) -> Set[str]: releases = set() try: - with urlopen(f'https://crates.io/api/v1/crates/{package_name}') as response: + with urlopen(f"https://crates.io/api/v1/crates/{package_name}") as response: response = json.load(response) - for version_info in response['versions']: - releases.add(version_info['num']) + for version_info in response["versions"]: + releases.add(version_info["num"]) except HTTPError as e: if e.code == 404: pass diff --git a/vulnerabilities/importers/safety_db.py b/vulnerabilities/importers/safety_db.py index 736af4364..72dd8e48a 100755 --- a/vulnerabilities/importers/safety_db.py +++ b/vulnerabilities/importers/safety_db.py @@ -42,21 +42,21 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import DataSource from vulnerabilities.data_source import DataSourceConfiguration +from vulnerabilities.data_source import Reference def validate_schema(advisory_dict): scheme = { - str: - [ - { - "advisory": str, - "cve": Or(None, Regex(r"CVE-\d+-\d+")), - "id": Regex(r"^pyup.io-\d"), - "specs": list, - "v": str - } - ] + str: [ + { + "advisory": str, + "cve": Or(None, Regex(r"CVE-\d+-\d+")), + "id": Regex(r"^pyup.io-\d"), + "specs": list, + "v": str, + } + ] } Schema(scheme).validate(advisory_dict) @@ -99,30 +99,33 @@ def updated_advisories(self) -> Set[Advisory]: for advisory in self._api_response[package_name]: impacted_purls, resolved_purls = categorize_versions( - package_name, all_package_versions, advisory['specs']) + package_name, all_package_versions, advisory["specs"] + ) - cve_ids = advisory.get('cve') or [''] + cve_ids = advisory.get("cve") or [""] # meaning if cve_ids is not [''] but either ['CVE-123'] or ['CVE-123, CVE-124'] if len(cve_ids[0]): - cve_ids = [s.strip() for s in cve_ids.split(',')] + cve_ids = [s.strip() for s in cve_ids.split(",")] + + reference = [Reference(reference_id=advisory["id"])] for cve_id in cve_ids: - advisories.append(Advisory( - cve_id=cve_id, - summary=advisory['advisory'], - reference_ids=[advisory['id']], - impacted_package_urls=impacted_purls, - resolved_package_urls=resolved_purls, - )) + advisories.append( + Advisory( + cve_id=cve_id, + summary=advisory["advisory"], + vuln_references=reference, + impacted_package_urls=impacted_purls, + resolved_package_urls=resolved_purls, + ) + ) return self.batch_advisories(advisories) def categorize_versions( - package_name: str, - all_versions: Set[str], - version_specs: Iterable[str], + package_name: str, all_versions: Set[str], version_specs: Iterable[str], ) -> Tuple[Set[PackageURL], Set[PackageURL]]: """ :return: impacted, resolved purls @@ -134,19 +137,11 @@ def categorize_versions( if any([version in r for r in ranges]): impacted_versions.add(version) - impacted_purls.add(PackageURL( - name=package_name, - type='pypi', - version=version, - )) + impacted_purls.add(PackageURL(name=package_name, type="pypi", version=version,)) resolved_purls = set() for version in all_versions - impacted_versions: - resolved_purls.add(PackageURL( - name=package_name, - type='pypi', - version=version - )) + resolved_purls.add(PackageURL(name=package_name, type="pypi", version=version)) return impacted_purls, resolved_purls @@ -161,9 +156,9 @@ def get(self, package_name: str) -> Set[str]: if package_name not in self.cache: releases = set() try: - with urlopen(f'https://pypi.org/pypi/{package_name}/json') as response: + with urlopen(f"https://pypi.org/pypi/{package_name}/json") as response: json_file = json.load(response) - releases = set(json_file['releases']) + releases = set(json_file["releases"]) except HTTPError as e: if e.code == 404: # PyPi does not have data about this package diff --git a/vulnerabilities/importers/ubuntu_usn.py b/vulnerabilities/importers/ubuntu_usn.py index bc52a285e..fe39850ad 100644 --- a/vulnerabilities/importers/ubuntu_usn.py +++ b/vulnerabilities/importers/ubuntu_usn.py @@ -29,6 +29,7 @@ from vulnerabilities.data_source import DataSource from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference @dataclasses.dataclass @@ -48,7 +49,7 @@ def updated_advisories(self): return self.batch_advisories(advisories) def create_etag(self, url): - etag = requests.head(url).headers.get('etag') + etag = requests.head(url).headers.get("etag") if not etag: return True @@ -63,33 +64,35 @@ def create_etag(self, url): def to_advisories(usn_db): advisories = [] for usn in usn_db: - usnid_data = get_usn_references(usn_db[usn]['id']) - for release in usn_db[usn]['releases']: - pkg_dict = usn_db[usn]['releases'][release] + reference = get_usn_references(usn_db[usn]["id"]) + for release in usn_db[usn]["releases"]: + pkg_dict = usn_db[usn]["releases"][release] safe_purls = get_purls(pkg_dict) - for cve in usn_db[usn].get('cves', ['']): + for cve in usn_db[usn].get("cves", [""]): # The db sometimes contains entries like # {'cves': ['python-pgsql vulnerabilities', 'CVE-2006-2313', 'CVE-2006-2314']} # This `if` filters entries like 'python-pgsql vulnerabilities' - if not cve.startswith('CVE-'): + if not cve.startswith("CVE-"): continue - advisories.append(Advisory( - cve_id=cve, - impacted_package_urls=[], - resolved_package_urls=safe_purls, - summary='', - reference_urls=usnid_data['reference_url'], - reference_ids=[usnid_data['reference_id']])) + advisories.append( + Advisory( + cve_id=cve, + impacted_package_urls=[], + resolved_package_urls=safe_purls, + summary="", + vuln_references=[reference], + ) + ) return advisories def get_usn_references(usn_id): - return {'reference_id': 'USN-' + usn_id, - 'reference_url': ['https://usn.ubuntu.com/{}/'.format(usn_id)] - } + return Reference( + reference_id="USN-" + usn_id, url="https://usn.ubuntu.com/{}/".format(usn_id) + ) def fetch(url): @@ -101,29 +104,21 @@ def fetch(url): def get_purls(pkg_dict): purls = set() - for pkg_name in pkg_dict.get('sources', []): - version = pkg_dict['sources'][pkg_name]['version'] + for pkg_name in pkg_dict.get("sources", []): + version = pkg_dict["sources"][pkg_name]["version"] # The db sometimes contains entries like {'postgresql': {'version': ''}} # This `if` ignores such entries if not version: continue - purls.add(PackageURL(name=pkg_name, - version=version, - type='deb', - namespace='ubuntu', - )) + purls.add(PackageURL(name=pkg_name, version=version, type="deb", namespace="ubuntu",)) - for pkg_name in pkg_dict['binaries']: - version = pkg_dict['binaries'][pkg_name]['version'] + for pkg_name in pkg_dict["binaries"]: + version = pkg_dict["binaries"][pkg_name]["version"] # The db sometimes contains entries like {'postgresql': {'version': ''}} # This `if` ignores such entries if not version: continue - purls.add(PackageURL(name=pkg_name, - version=version, - type='deb', - namespace='ubuntu', - )) + purls.add(PackageURL(name=pkg_name, version=version, type="deb", namespace="ubuntu",)) return purls diff --git a/vulnerabilities/tests/test_archlinux.py b/vulnerabilities/tests/test_archlinux.py index 7a271dd39..83148ab63 100644 --- a/vulnerabilities/tests/test_archlinux.py +++ b/vulnerabilities/tests/test_archlinux.py @@ -63,9 +63,8 @@ def test_import(self): return_value=self.mock_response ): runner.run() - assert models.Vulnerability.objects.count() == 6 - assert models.VulnerabilityReference.objects.count() == 4 + assert models.VulnerabilityReference.objects.count() == 10 assert models.PackageRelatedVulnerability.objects.filter( is_vulnerable=True).count() == 12 assert models.PackageRelatedVulnerability.objects.filter( diff --git a/vulnerabilities/tests/test_debian_oval.py b/vulnerabilities/tests/test_debian_oval.py index 51c5408ae..291da20e9 100644 --- a/vulnerabilities/tests/test_debian_oval.py +++ b/vulnerabilities/tests/test_debian_oval.py @@ -70,8 +70,6 @@ def test_get_data_from_xml_doc(self, mock_write): version='2.3.9', qualifiers=OrderedDict([('distro', 'wheezy')]), subpath=None)}, - reference_urls=set(), - reference_ids=[], cve_id='CVE-2002-2443' ), Advisory( @@ -98,8 +96,6 @@ def test_get_data_from_xml_doc(self, mock_write): version='0:1.11.1+dfsg-5+deb7u1', qualifiers=OrderedDict([('distro', 'wheezy')]), subpath=None)}, - reference_urls=set(), - reference_ids=[], cve_id='CVE-2001-1593') } diff --git a/vulnerabilities/tests/test_gentoo.py b/vulnerabilities/tests/test_gentoo.py index 8385f1750..6659e4a77 100644 --- a/vulnerabilities/tests/test_gentoo.py +++ b/vulnerabilities/tests/test_gentoo.py @@ -30,6 +30,7 @@ from vulnerabilities.importers.gentoo import GentooDataSource from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEST_DATA = os.path.join(BASE_DIR, "test_data/gentoo/glsa-201709-09.xml") @@ -88,9 +89,10 @@ def test_process_file(self): exp_data = [ Advisory( - summary=('A command injection vulnerability in ' - 'Subversion may allow remote\n ' - 'attackers to execute arbitrary code.\n '), + summary=( + 'A command injection vulnerability in ' + 'Subversion may allow remote\n ' + 'attackers to execute arbitrary code.\n '), impacted_package_urls={ PackageURL( type='ebuild', @@ -107,8 +109,10 @@ def test_process_file(self): version='1.9.7', qualifiers=OrderedDict(), subpath=None)}, - reference_urls=[], - reference_ids=['GLSA-201709-09'], + vuln_references=[ + Reference( + url='https://security.gentoo.org/glsa/201709-09', + reference_id='GLSA-201709-09')], cve_id='CVE-2017-9800')] found_data = self.data_src.process_file(TEST_DATA) diff --git a/vulnerabilities/tests/test_github.py b/vulnerabilities/tests/test_github.py index eb65c359d..f6e0c27c2 100644 --- a/vulnerabilities/tests/test_github.py +++ b/vulnerabilities/tests/test_github.py @@ -32,13 +32,14 @@ from requests.models import Response from packageurl import PackageURL +from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference from vulnerabilities.importers.github import GitHubAPIDataSource from vulnerabilities.importers.github import MavenVersionAPI from vulnerabilities.importers.github import ComposerVersionAPI from vulnerabilities.importers.github import NugetVersionAPI from vulnerabilities.importers.github import GitHubTokenError from vulnerabilities.importers.github import query -from vulnerabilities.data_source import Advisory BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEST_DATA = os.path.join(BASE_DIR, 'test_data') @@ -186,8 +187,9 @@ def test_process_response(self): subpath=None, ), }, - reference_urls=[], - reference_ids={'GHSA-qcxh-w3j9-58qr'}, + vuln_references=[Reference( + url='https://github.com/advisories/GHSA-qcxh-w3j9-58qr', + reference_id='GHSA-qcxh-w3j9-58qr')], cve_id='CVE-2019-0199', ), Advisory( @@ -212,8 +214,9 @@ def test_process_response(self): subpath=None, ) }, - reference_urls=[], - reference_ids={'GHSA-qcxh-w3j9-58qr'}, + vuln_references=[Reference( + url='https://github.com/advisories/GHSA-qcxh-w3j9-58qr', + reference_id='GHSA-qcxh-w3j9-58qr')], cve_id='CVE-2019-0199', ), Advisory( @@ -237,8 +240,9 @@ def test_process_response(self): subpath=None, ), }, - reference_urls=[], - reference_ids={'GHSA-c9hw-wf7x-jp9j'}, + vuln_references=[Reference( + url='https://github.com/advisories/GHSA-c9hw-wf7x-jp9j', + reference_id='GHSA-c9hw-wf7x-jp9j')], cve_id='CVE-2020-1938', ), Advisory( @@ -262,8 +266,9 @@ def test_process_response(self): subpath=None, ), }, - reference_urls=[], - reference_ids={'GHSA-c9hw-wf7x-jp9j'}, + vuln_references=[Reference( + url='https://github.com/advisories/GHSA-c9hw-wf7x-jp9j', + reference_id='GHSA-c9hw-wf7x-jp9j')], cve_id='CVE-2020-1938', ), Advisory( @@ -288,8 +293,9 @@ def test_process_response(self): subpath=None, ) }, - reference_urls=[], - reference_ids={'GHSA-c9hw-wf7x-jp9j'}, + vuln_references=[Reference( + url='https://github.com/advisories/GHSA-c9hw-wf7x-jp9j', + reference_id='GHSA-c9hw-wf7x-jp9j')], cve_id='CVE-2020-1938', ), ] @@ -433,7 +439,8 @@ def test_artifact_url(self): def test_extract_versions(self): expected_versions = {'1.2.2', '1.2.3', '1.3.0'} - assert expected_versions == self.version_api.extract_versions(self.response) + assert expected_versions == self.version_api.extract_versions( + self.response) def test_load_to_api(self): diff --git a/vulnerabilities/tests/test_import_runner.py b/vulnerabilities/tests/test_import_runner.py index 2ab0dc46c..5af1975b1 100644 --- a/vulnerabilities/tests/test_import_runner.py +++ b/vulnerabilities/tests/test_import_runner.py @@ -28,6 +28,7 @@ from vulnerabilities.data_source import Advisory from vulnerabilities.data_source import DataSource from vulnerabilities.data_source import PackageURL +from vulnerabilities.data_source import Reference from vulnerabilities.import_runner import ImportRunner from vulnerabilities.import_runner import _insert_vulnerabilities_and_references @@ -70,11 +71,20 @@ def save(self): Advisory( cve_id='MOCK-CVE-2020-1337', summary='vulnerability description here', - reference_urls=['https://example.com/with/more/info/MOCK-CVE-2020-1337'], - impacted_package_urls=[PackageURL(name='mock-webserver', type='pypi', version='1.2.33')], - resolved_package_urls=[PackageURL(name='mock-webserver', type='pypi', version='1.2.34')], - ) -] + vuln_references=[ + Reference( + url='https://example.com/with/more/info/MOCK-CVE-2020-1337')], + impacted_package_urls=[ + PackageURL( + name='mock-webserver', + type='pypi', + version='1.2.33')], + resolved_package_urls=[ + PackageURL( + name='mock-webserver', + type='pypi', + version='1.2.34')], + )] def make_import_runner(added_advs=None, updated_advs=None): diff --git a/vulnerabilities/tests/test_openssl.py b/vulnerabilities/tests/test_openssl.py index 178794281..23ebb813e 100644 --- a/vulnerabilities/tests/test_openssl.py +++ b/vulnerabilities/tests/test_openssl.py @@ -28,6 +28,7 @@ from vulnerabilities.importers.openssl import OpenSSLDataSource from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEST_DATA = os.path.join(BASE_DIR, 'test_data/', 'openssl_xml_data.xml') @@ -86,10 +87,9 @@ def test_to_advisory(self): version='1.1.1g', qualifiers=OrderedDict(), subpath=None)}, - reference_urls=[ - 'https://github.com/openssl/openssl/commit/' + - 'eb563247aef3e83dda7679c43f9649270462e5b1'], - reference_ids=[], + vuln_references=[Reference( + url='https://github.com/openssl/openssl/commit/' + + 'eb563247aef3e83dda7679c43f9649270462e5b1')], cve_id='CVE-2020-1967'), Advisory( summary='There is an overflow bug in the x64_64 Montgomery squaring procedure ' @@ -189,11 +189,12 @@ def test_to_advisory(self): qualifiers=OrderedDict(), subpath=None), PackageURL(type='generic', namespace=None, name='openssl', version='1.0.2u', qualifiers=OrderedDict(), subpath=None)}, - reference_urls=['https://github.com/openssl/openssl/commit/' + - '419102400a2811582a7a3d4a4e317d72e5ce0a8f', - 'https://github.com/openssl/openssl/commit/' + - 'f1c5eea8a817075d31e43f5876993c6710238c98'], - reference_ids=[], + vuln_references=[Reference( + url='https://github.com/openssl/openssl/commit/' + + '419102400a2811582a7a3d4a4e317d72e5ce0a8f'), + Reference( + url='https://github.com/openssl/openssl/commit/' + + 'f1c5eea8a817075d31e43f5876993c6710238c98')], cve_id='CVE-2019-1551') ] found_data = OpenSSLDataSource.to_advisories(data) diff --git a/vulnerabilities/tests/test_redhat_importer.py b/vulnerabilities/tests/test_redhat_importer.py index 22a8252fc..8447c508a 100644 --- a/vulnerabilities/tests/test_redhat_importer.py +++ b/vulnerabilities/tests/test_redhat_importer.py @@ -29,9 +29,10 @@ import vulnerabilities.importers.redhat as redhat from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -TEST_DATA = os.path.join(BASE_DIR, 'test_data/', 'redhat.json') +TEST_DATA = os.path.join(BASE_DIR, "test_data/", "redhat.json") def load_test_data(): @@ -40,92 +41,156 @@ def load_test_data(): class TestRedhat(unittest.TestCase): - def test_rpm_to_purl(self): assert redhat.rpm_to_purl("foobar") is None assert redhat.rpm_to_purl("foo-bar-devel-0:sys76") is None assert redhat.rpm_to_purl("kernel-0:2.6.32-754.el6") == PackageURL( - type='rpm', - namespace='redhat', - name='kernel', - version='2.6.32-754.el6', + type="rpm", + namespace="redhat", + name="kernel", + version="2.6.32-754.el6", qualifiers=OrderedDict(), - subpath=None) + subpath=None, + ) def test_to_advisory(self): data = load_test_data() expected_data = { Advisory( - summary='CVE-2016-9401 bash: popd controlled free', + summary="CVE-2016-9401 bash: popd controlled free", impacted_package_urls=[ PackageURL( - type='rpm', - namespace='redhat', - name='bash', - version='4.2.46-28.el7', + type="rpm", + namespace="redhat", + name="bash", + version="4.2.46-28.el7", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='rpm', - namespace='redhat', - name='bash', - version='4.1.2-48.el6', + type="rpm", + namespace="redhat", + name="bash", + version="4.1.2-48.el6", qualifiers=OrderedDict(), - subpath=None)], + subpath=None, + ), + ], resolved_package_urls=[], - reference_urls=[ - 'https://access.redhat.com/hydra/rest/securitydata/cve/CVE-2016-9401.json'], - reference_ids=[ - 'RHSA-2017:1931', - 'RHSA-2017:0725'], - cve_id='CVE-2016-9401'), + vuln_references=sorted( + [ + Reference( + url="https://bugzilla.redhat.com/show_bug.cgi?id=1396383", + reference_id="1396383", + ), + Reference( + url="https://access.redhat.com/errata/RHSA-2017:1931", + reference_id="RHSA-2017:1931", + ), + Reference( + url="https://access.redhat.com/errata/RHSA-2017:0725", + reference_id="RHSA-2017:0725", + ), + Reference( + url="https://access.redhat.com/hydra/rest/securitydata/cve/CVE-2016-9401.json", # nopep8 + reference_id="", + ), + ], + key=lambda x: x.url, + ), + cve_id="CVE-2016-9401", + ), Advisory( - summary=('CVE-2016-10200 kernel: l2tp: Race condition ' - 'in the L2TPv3 IP encapsulation feature'), + summary=( + "CVE-2016-10200 kernel: l2tp: Race condition " + "in the L2TPv3 IP encapsulation feature" + ), impacted_package_urls=[ PackageURL( - type='rpm', - namespace='redhat', - name='kernel-rt', - version='3.10.0-693.rt56.617.el7', + type="rpm", + namespace="redhat", + name="kernel-rt", + version="3.10.0-693.rt56.617.el7", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='rpm', - namespace='redhat', - name='kernel', - version='3.10.0-693.el7', + type="rpm", + namespace="redhat", + name="kernel", + version="3.10.0-693.el7", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='rpm', - namespace='redhat', - name='kernel', - version='3.10.0-514.28.1.el7', + type="rpm", + namespace="redhat", + name="kernel", + version="3.10.0-514.28.1.el7", qualifiers=OrderedDict(), - subpath=None)], + subpath=None, + ), + ], resolved_package_urls=[], - reference_urls=[ - 'https://access.redhat.com/hydra/rest/securitydata/cve/CVE-2016-10200.json'], - reference_ids=[ - 'RHSA-2017:1842', - 'RHSA-2017:2437', - 'RHSA-2017:2077', - 'RHSA-2017:2444'], - cve_id='CVE-2016-10200'), + vuln_references=sorted( + [ + Reference( + url="https://bugzilla.redhat.com/show_bug.cgi?id=1430347", + reference_id="1430347", + ), + Reference( + url="https://access.redhat.com/errata/RHSA-2017:1842", + reference_id="RHSA-2017:1842", + ), + Reference( + url="https://access.redhat.com/errata/RHSA-2017:2437", + reference_id="RHSA-2017:2437", + ), + Reference( + url="https://access.redhat.com/errata/RHSA-2017:2077", + reference_id="RHSA-2017:2077", + ), + Reference( + url="https://access.redhat.com/errata/RHSA-2017:2444", + reference_id="RHSA-2017:2444", + ), + Reference( + url="https://access.redhat.com/hydra/rest/securitydata/cve/CVE-2016-10200.json", # nopep8 + reference_id="", + ), + ], + key=lambda x: x.url, + ), + cve_id="CVE-2016-10200", + ), Advisory( - summary=('CVE-2017-12168 Kernel: kvm: ARM64: ' - 'assert failure when accessing PMCCNTR register'), + summary=( + "CVE-2017-12168 Kernel: kvm: ARM64: " + "assert failure when accessing PMCCNTR register" + ), impacted_package_urls=[], resolved_package_urls=[], - reference_urls=[ - 'https://access.redhat.com/hydra/rest/securitydata/cve/CVE-2017-12168.json'], - reference_ids=[], - cve_id='CVE-2017-12168'), + vuln_references=sorted( + [ + Reference( + url="https://bugzilla.redhat.com/show_bug.cgi?id=1492984", + reference_id="1492984", + ), + Reference( + url="https://access.redhat.com/hydra/rest/securitydata/cve/CVE-2017-12168.json", # nopep8 + reference_id="", + ), + ], + key=lambda x: x.url, + ), + cve_id="CVE-2017-12168", + ), } found_data = set() for adv in data: - found_data.add(redhat.to_advisory(adv)) + adv = redhat.to_advisory(adv) + adv.vuln_references = sorted(adv.vuln_references, key=lambda x: x.url) + found_data.add(adv) assert expected_data == found_data diff --git a/vulnerabilities/tests/test_retiredotnet.py b/vulnerabilities/tests/test_retiredotnet.py index 9ea38e297..fa29d7899 100644 --- a/vulnerabilities/tests/test_retiredotnet.py +++ b/vulnerabilities/tests/test_retiredotnet.py @@ -28,16 +28,17 @@ from vulnerabilities.importers.retiredotnet import RetireDotnetDataSource from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) class TestRetireDotnetDataSource(TestCase): - @classmethod def setUpClass(cls): data_source_cfg = { - 'repository_url': 'https://test.net', } + "repository_url": "https://test.net", + } cls.data_src = RetireDotnetDataSource(1, config=data_source_cfg) def test_vuln_id_from_desc(self): @@ -54,69 +55,85 @@ def test_process_file(self): path = os.path.join(BASE_DIR, "test_data/retiredotnet/test_file.json") expected_data = Advisory( - summary=('Microsoft Security Advisory CVE-2019-0982: ' - 'ASP.NET Core Denial of Service Vulnerability'), + summary=( + "Microsoft Security Advisory CVE-2019-0982: " + "ASP.NET Core Denial of Service Vulnerability" + ), impacted_package_urls={ PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.0.4', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.0.4", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.0.0', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.0.0", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.0.2', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.0.2", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.0.3', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.0.3", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.1.0', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.1.0", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.0.1', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.0.1", qualifiers=OrderedDict(), - subpath=None)}, + subpath=None, + ), + }, resolved_package_urls={ PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.0.11', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.0.11", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='nuget', + type="nuget", namespace=None, - name='Microsoft.AspNetCore.SignalR.Protocols.MessagePack', - version='1.1.5', + name="Microsoft.AspNetCore.SignalR.Protocols.MessagePack", + version="1.1.5", qualifiers=OrderedDict(), - subpath=None)}, - reference_urls=['https://github.com/aspnet/Announcements/issues/359'], - reference_ids=[], - cve_id='CVE-2019-0982') + subpath=None, + ), + }, + vuln_references=[ + Reference( + url="https://github.com/aspnet/Announcements/issues/359", reference_id="" + ) + ], + cve_id="CVE-2019-0982", + ) found_data = self.data_src.process_file(path) diff --git a/vulnerabilities/tests/test_ruby.py b/vulnerabilities/tests/test_ruby.py index 8e3e98044..0e5ec5969 100644 --- a/vulnerabilities/tests/test_ruby.py +++ b/vulnerabilities/tests/test_ruby.py @@ -31,90 +31,112 @@ from vulnerabilities.importers.ruby import RubyDataSource from vulnerabilities.data_source import GitDataSourceConfiguration from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -TEST_DATA = os.path.join(BASE_DIR, 'test_data', 'ruby') +TEST_DATA = os.path.join(BASE_DIR, "test_data", "ruby") MOCK_ADDED_FILES = [] -for filepath in pathlib.Path(TEST_DATA).glob('**/*.yml'): +for filepath in pathlib.Path(TEST_DATA).glob("**/*.yml"): MOCK_ADDED_FILES.append(filepath.absolute()) class RubyDataSourceTest(TestCase): - @classmethod def setUpClass(cls): data_source_cfg = { - 'repository_url': 'https://github.com/rubysec/ruby-advisory-db.git', } + "repository_url": "https://github.com/rubysec/ruby-advisory-db.git", + } cls.data_src = RubyDataSource(1, config=data_source_cfg) - @patch('vulnerabilities.importers.ruby.rubyAPI.get_all_version_of_package', - return_value={'1.0.0', '1.8.0', '2.0.3'}) + @patch( + "vulnerabilities.importers.ruby.rubyAPI.get_all_version_of_package", + return_value={"1.0.0", "1.8.0", "2.0.3"}, + ) def test_process_file(self, mock_write): expected_advisories = { Advisory( - summary=('An issue was discovered in' - ' rack-protection/lib/rack/protection/path_traversal.rb\n' - 'in Sinatra 2.x before 2.0.1 on Windows.' - ' Path traversal is possible via backslash\ncharacters.\n'), + summary=( + "An issue was discovered in" + " rack-protection/lib/rack/protection/path_traversal.rb\n" + "in Sinatra 2.x before 2.0.1 on Windows." + " Path traversal is possible via backslash\ncharacters.\n" + ), impacted_package_urls={ PackageURL( - type='gem', + type="gem", namespace=None, - name='sinatra', - version='1.8.0', + name="sinatra", + version="1.8.0", qualifiers=OrderedDict(), - subpath=None)}, + subpath=None, + ) + }, resolved_package_urls={ PackageURL( - type='gem', + type="gem", namespace=None, - name='sinatra', - version='1.0.0', + name="sinatra", + version="1.0.0", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='gem', + type="gem", namespace=None, - name='sinatra', - version='2.0.3', + name="sinatra", + version="2.0.3", qualifiers=OrderedDict(), - subpath=None)}, - reference_urls=['https://github.com/sinatra/sinatra/pull/1379'], - reference_ids=[], - cve_id='CVE-2018-7212'), + subpath=None, + ), + }, + vuln_references=[ + Reference(url="https://github.com/sinatra/sinatra/pull/1379") + ], + cve_id="CVE-2018-7212", + ), Advisory( - summary=('Sinatra before 2.0.2 has XSS via the 400 Bad Request ' - 'page that occurs upon a params parser exception.\n'), + summary=( + "Sinatra before 2.0.2 has XSS via the 400 Bad Request " + "page that occurs upon a params parser exception.\n" + ), impacted_package_urls={ PackageURL( - type='gem', + type="gem", namespace=None, - name='sinatra', - version='1.0.0', + name="sinatra", + version="1.0.0", qualifiers=OrderedDict(), - subpath=None), + subpath=None, + ), PackageURL( - type='gem', + type="gem", namespace=None, - name='sinatra', - version='1.8.0', + name="sinatra", + version="1.8.0", qualifiers=OrderedDict(), - subpath=None)}, + subpath=None, + ), + }, resolved_package_urls={ PackageURL( - type='gem', + type="gem", namespace=None, - name='sinatra', - version='2.0.3', + name="sinatra", + version="2.0.3", qualifiers=OrderedDict(), - subpath=None)}, - reference_urls=['https://github.com/sinatra/sinatra/issues/1428'], - reference_ids=[], - cve_id='CVE-2018-11627'), - None} + subpath=None, + ) + }, + vuln_references=[ + Reference(url="https://github.com/sinatra/sinatra/issues/1428") + ], + cve_id="CVE-2018-11627", + ), + None, + } found_advisories = set() @@ -124,13 +146,12 @@ def test_process_file(self, mock_write): def test_categorize_versions(self): - all_versions = {'1.0.0', '1.2.0', '9.0.2', '0.2.3'} - safe_ver_ranges = ['==1.0.0', '>1.2.0'] + all_versions = {"1.0.0", "1.2.0", "9.0.2", "0.2.3"} + safe_ver_ranges = ["==1.0.0", ">1.2.0"] - exp_safe_vers = {'1.0.0', '9.0.2'} - exp_aff_vers = {'1.2.0', '0.2.3'} + exp_safe_vers = {"1.0.0", "9.0.2"} + exp_aff_vers = {"1.2.0", "0.2.3"} - safe_vers, aff_vers = self.data_src.categorize_versions( - all_versions, safe_ver_ranges) + safe_vers, aff_vers = self.data_src.categorize_versions(all_versions, safe_ver_ranges) assert exp_aff_vers == aff_vers assert exp_safe_vers == safe_vers diff --git a/vulnerabilities/tests/test_safety_db.py b/vulnerabilities/tests/test_safety_db.py index 110afca63..e0220846f 100644 --- a/vulnerabilities/tests/test_safety_db.py +++ b/vulnerabilities/tests/test_safety_db.py @@ -30,6 +30,7 @@ from vulnerabilities.import_runner import ImportRunner from vulnerabilities.importers.safety_db import VersionAPI from vulnerabilities.importers.safety_db import categorize_versions +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEST_DATA = os.path.join(BASE_DIR, 'test_data/') diff --git a/vulnerabilities/tests/test_suse_backports.py b/vulnerabilities/tests/test_suse_backports.py index 4a4e58d98..fa36c9534 100644 --- a/vulnerabilities/tests/test_suse_backports.py +++ b/vulnerabilities/tests/test_suse_backports.py @@ -65,8 +65,6 @@ def test_process_file(self): version='3.0.10-1.1.1', qualifiers=OrderedDict(), subpath=None)], - reference_urls=[], - reference_ids=[], cve_id='CVE-2009-1313'), Advisory( summary='', @@ -79,8 +77,6 @@ def test_process_file(self): version='3.5-1.1.5', qualifiers=OrderedDict(), subpath=None)], - reference_urls=[], - reference_ids=[], cve_id='CVE-2009-1313'), Advisory( summary='', @@ -93,8 +89,6 @@ def test_process_file(self): version='3.0.10-1.1.1', qualifiers=OrderedDict(), subpath=None)], - reference_urls=[], - reference_ids=[], cve_id='CVE-2009-1313'), Advisory( summary='', @@ -107,8 +101,6 @@ def test_process_file(self): version='0.7.0.r4359-15.9.2', qualifiers=OrderedDict(), subpath=None)], - reference_urls=[], - reference_ids=[], cve_id='CVE-2009-0365'), Advisory( summary='', @@ -121,8 +113,6 @@ def test_process_file(self): version='0.7.0.r4359-15.9.2', qualifiers=OrderedDict(), subpath=None)], - reference_urls=[], - reference_ids=[], cve_id='CVE-2009-0578'), ] diff --git a/vulnerabilities/tests/test_ubuntu.py b/vulnerabilities/tests/test_ubuntu.py index 14aa69bc9..b12ff29c9 100644 --- a/vulnerabilities/tests/test_ubuntu.py +++ b/vulnerabilities/tests/test_ubuntu.py @@ -12,6 +12,7 @@ from vulnerabilities.oval_parser import OvalParser from vulnerabilities.importers.ubuntu import UbuntuDataSource from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEST_DATA = os.path.join(BASE_DIR, "test_data/") @@ -224,14 +225,13 @@ def test_get_data_from_xml_doc(self, mock_write): version='2.14-2', qualifiers=OrderedDict(), subpath=None)}, - reference_urls={ - 'http://www.openwall.com/lists/oss-security/2016/10/18/11', - 'https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2016-8860', - 'http://people.canonical.com/~ubuntu-security/cve/2016/CVE-2016-8860.html', - 'https://github.com/torproject/tor/commit/3cea86eb2fbb65949673eb4ba8ebb695c87a57ce', - 'https://blog.torproject.org/blog/tor-0289-released-important-fixes', - 'https://trac.torproject.org/projects/tor/ticket/20384'}, - reference_ids=[], + vuln_references=sorted([ + Reference(url='http://www.openwall.com/lists/oss-security/2016/10/18/11'), + Reference(url='https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2016-8860'), + Reference(url='http://people.canonical.com/~ubuntu-security/cve/2016/CVE-2016-8860.html'), + Reference(url='https://github.com/torproject/tor/commit/3cea86eb2fbb65949673eb4ba8ebb695c87a57ce'), + Reference(url='https://blog.torproject.org/blog/tor-0289-released-important-fixes'), + Reference(url='https://trac.torproject.org/projects/tor/ticket/20384')],key=lambda x : x.url), cve_id='CVE-2016-8860'), Advisory( summary=('Heap-based buffer overflow in the bm_readbody_bmp function' @@ -262,11 +262,10 @@ def test_get_data_from_xml_doc(self, mock_write): version='2.14-2', qualifiers=OrderedDict(), subpath=None)}, - reference_urls={ - 'http://people.canonical.com/~ubuntu-security/cve/2016/CVE-2016-8703.html', - 'https://blogs.gentoo.org/ago/2016/08/08/potrace-multiplesix-heap-based-buffer-overflow-in-bm_readbody_bmp-bitmap_io-c/', - 'https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2016-8703'}, - reference_ids=[], + vuln_references=sorted([ + Reference(url='http://people.canonical.com/~ubuntu-security/cve/2016/CVE-2016-8703.html'), + Reference(url='https://blogs.gentoo.org/ago/2016/08/08/potrace-multiplesix-heap-based-buffer-overflow-in-bm_readbody_bmp-bitmap_io-c/'), + Reference(url='https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2016-8703')],key=lambda x: x.url), cve_id='CVE-2016-8703')} xml_doc = ET.parse(os.path.join(TEST_DATA, "ubuntu_oval_data.xml")) @@ -274,6 +273,8 @@ def test_get_data_from_xml_doc(self, mock_write): with patch('vulnerabilities.importers.ubuntu.UbuntuDataSource.batch_advisories', new=return_adv): data = {i for i in self.ubuntu_data_src.get_data_from_xml_doc(xml_doc,{"type":"deb"})} + for adv in data : + adv.vuln_references = sorted(adv.vuln_references, key=lambda x : x.url) assert expected_data == data def test_create_etag(self): diff --git a/vulnerabilities/tests/test_ubuntu_usn.py b/vulnerabilities/tests/test_ubuntu_usn.py index 9a520a408..7e155a654 100644 --- a/vulnerabilities/tests/test_ubuntu_usn.py +++ b/vulnerabilities/tests/test_ubuntu_usn.py @@ -31,28 +31,28 @@ from packageurl import PackageURL from vulnerabilities.data_source import Advisory +from vulnerabilities.data_source import Reference import vulnerabilities.importers.ubuntu_usn as ubuntu_usn BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -TEST_DATA = os.path.join(BASE_DIR, 'test_data/', 'ubuntu_usn_db', 'database-all.json.bz2') +TEST_DATA = os.path.join(BASE_DIR, "test_data/", "ubuntu_usn_db", "database-all.json.bz2") class TestUbuntuUSNDataSource(TestCase): @classmethod def setUpClass(cls): - data_src_cfg = {'etags': {}, 'db_url': 'http://exampledb.com'} + data_src_cfg = {"etags": {}, "db_url": "http://exampledb.com"} cls.data_src = ubuntu_usn.UbuntuUSNDataSource(batch_size=1, config=data_src_cfg) - with open(TEST_DATA, 'rb') as f: + with open(TEST_DATA, "rb") as f: cls.raw_data = f.read() cls.db = json.loads(bz2.decompress(cls.raw_data)) def test_get_usn_references(self): - eg_usn = '435-1' - expected_references = { - 'reference_id': 'USN-435-1', - 'reference_url': ['https://usn.ubuntu.com/435-1/'], - } + eg_usn = "435-1" + expected_references = Reference( + reference_id="USN-435-1", url="https://usn.ubuntu.com/435-1/" + ) found_references = ubuntu_usn.get_usn_references(eg_usn) assert found_references == expected_references @@ -61,65 +61,65 @@ def test_fetch(self): mock_response = MagicMock() mock_response.content = self.raw_data - with patch('vulnerabilities.importers.ubuntu_usn.requests.get', return_value=mock_response): - assert ubuntu_usn.fetch('www.db.com') == self.db + with patch("vulnerabilities.importers.ubuntu_usn.requests.get", return_value=mock_response): + assert ubuntu_usn.fetch("www.db.com") == self.db def test_get_purls(self): - eg_pkg_dict_1 = self.db['763-1']['releases']['hardy'] - eg_pkg_dict_2 = self.db['763-1']['releases']['dapper'] - eg_pkg_dict_3 = self.db['763-1']['releases']['intrepid'] + eg_pkg_dict_1 = self.db["763-1"]["releases"]["hardy"] + eg_pkg_dict_2 = self.db["763-1"]["releases"]["dapper"] + eg_pkg_dict_3 = self.db["763-1"]["releases"]["intrepid"] exp_pkgs_1 = { PackageURL( - type='deb', - namespace='ubuntu', - name='xine-lib', - version='1.1.11.1-1ubuntu3.4', + type="deb", + namespace="ubuntu", + name="xine-lib", + version="1.1.11.1-1ubuntu3.4", qualifiers=OrderedDict(), subpath=None, ), PackageURL( - type='deb', - namespace='ubuntu', - name='libxine1', - version='1.1.11.1-1ubuntu3.4', + type="deb", + namespace="ubuntu", + name="libxine1", + version="1.1.11.1-1ubuntu3.4", qualifiers=OrderedDict(), subpath=None, ), } exp_pkgs_2 = { PackageURL( - type='deb', - namespace='ubuntu', - name='libxine-main1', - version='1.1.1+ubuntu2-7.12', + type="deb", + namespace="ubuntu", + name="libxine-main1", + version="1.1.1+ubuntu2-7.12", qualifiers=OrderedDict(), subpath=None, ), PackageURL( - type='deb', - namespace='ubuntu', - name='xine-lib', - version='1.1.1+ubuntu2-7.12', + type="deb", + namespace="ubuntu", + name="xine-lib", + version="1.1.1+ubuntu2-7.12", qualifiers=OrderedDict(), subpath=None, ), } exp_pkgs_3 = { PackageURL( - type='deb', - namespace='ubuntu', - name='xine-lib', - version='1.1.15-0ubuntu3.3', + type="deb", + namespace="ubuntu", + name="xine-lib", + version="1.1.15-0ubuntu3.3", qualifiers=OrderedDict(), subpath=None, ), PackageURL( - type='deb', - namespace='ubuntu', - name='libxine1', - version='1.1.15-0ubuntu3.3', + type="deb", + namespace="ubuntu", + name="libxine1", + version="1.1.15-0ubuntu3.3", qualifiers=OrderedDict(), subpath=None, ), @@ -133,54 +133,60 @@ def test_to_advisories(self): expected_advisories = { Advisory( - summary='', + summary="", impacted_package_urls=[], resolved_package_urls={ PackageURL( - type='deb', - namespace='ubuntu', - name='xine-lib', - version='1.1.15-0ubuntu3.3', + type="deb", + namespace="ubuntu", + name="xine-lib", + version="1.1.15-0ubuntu3.3", qualifiers=OrderedDict(), subpath=None, ), PackageURL( - type='deb', - namespace='ubuntu', - name='libxine1', - version='1.1.15-0ubuntu3.3', + type="deb", + namespace="ubuntu", + name="libxine1", + version="1.1.15-0ubuntu3.3", qualifiers=OrderedDict(), subpath=None, ), }, - reference_urls=['https://usn.ubuntu.com/763-1/'], - reference_ids=['USN-763-1'], - cve_id='CVE-2009-0698', + vuln_references=[ + Reference( + url="https://usn.ubuntu.com/763-1/", reference_id="USN-763-1" + ) + ], + cve_id="CVE-2009-0698", ), Advisory( - summary='', + summary="", impacted_package_urls=[], resolved_package_urls={ PackageURL( - type='deb', - namespace='ubuntu', - name='xine-lib', - version='1.1.15-0ubuntu3.3', + type="deb", + namespace="ubuntu", + name="xine-lib", + version="1.1.15-0ubuntu3.3", qualifiers=OrderedDict(), subpath=None, ), PackageURL( - type='deb', - namespace='ubuntu', - name='libxine1', - version='1.1.15-0ubuntu3.3', + type="deb", + namespace="ubuntu", + name="libxine1", + version="1.1.15-0ubuntu3.3", qualifiers=OrderedDict(), subpath=None, ), }, - reference_urls=['https://usn.ubuntu.com/763-1/'], - reference_ids=['USN-763-1'], - cve_id='CVE-2009-1274', + vuln_references=[ + Reference( + url="https://usn.ubuntu.com/763-1/", reference_id="USN-763-1" + ) + ], + cve_id="CVE-2009-1274", ), } found_advisories = set(self.data_src.to_advisories(self.db)) @@ -191,9 +197,9 @@ def test_create_etag(self): assert self.data_src.config.etags == {} mock_response = MagicMock() - mock_response.headers = {'etag': '2131151243&2191'} + mock_response.headers = {"etag": "2131151243&2191"} - with patch('vulnerabilities.importers.ubuntu.requests.head', return_value=mock_response): - assert self.data_src.create_etag('https://example.org') - assert self.data_src.config.etags == {'https://example.org': '2131151243&2191'} - assert not self.data_src.create_etag('https://example.org') + with patch("vulnerabilities.importers.ubuntu.requests.head", return_value=mock_response): + assert self.data_src.create_etag("https://example.org") + assert self.data_src.config.etags == {"https://example.org": "2131151243&2191"} + assert not self.data_src.create_etag("https://example.org")