|
20 | 20 | # for any legal advice. |
21 | 21 | # VulnerableCode is a free software code scanning tool from nexB Inc. and others. |
22 | 22 | # Visit https://github.com/nexB/vulnerablecode/ for support and download. |
23 | | -import logging |
| 23 | + |
24 | 24 | from typing import Any |
25 | 25 | from typing import Iterable |
26 | 26 | from typing import List |
27 | 27 | from typing import Mapping |
28 | 28 | from typing import Set |
29 | 29 |
|
30 | | -import yaml |
31 | 30 | from packageurl import PackageURL |
32 | 31 | from schema import Or |
33 | 32 | from schema import Regex |
|
36 | 35 | from vulnerabilities.data_source import Advisory |
37 | 36 | from vulnerabilities.data_source import GitDataSource |
38 | 37 | from vulnerabilities.data_source import Reference |
| 38 | +from vulnerabilities.helpers import load_yaml |
39 | 39 |
|
40 | 40 |
|
41 | 41 | def validate_schema(advisory_dict): |
@@ -91,19 +91,17 @@ def updated_advisories(self) -> Set[Advisory]: |
91 | 91 | def _process_file(self, path) -> List[Advisory]: |
92 | 92 | advisories = [] |
93 | 93 |
|
94 | | - with open(path) as f: |
95 | | - record = yaml.safe_load(f) |
96 | | - |
97 | | - if record["packages"] is None: |
98 | | - return advisories |
99 | | - validate_schema(record) |
| 94 | + record = load_yaml(path) |
| 95 | + if record["packages"] is None: |
| 96 | + return advisories |
| 97 | + validate_schema(record) |
100 | 98 |
|
101 | | - for p in record["packages"]: |
102 | | - advisories.extend( |
103 | | - self._load_advisories( |
104 | | - p["pkg"], record["distroversion"], record["reponame"], record["archs"], |
105 | | - ) |
| 99 | + for p in record["packages"]: |
| 100 | + advisories.extend( |
| 101 | + self._load_advisories( |
| 102 | + p["pkg"], record["distroversion"], record["reponame"], record["archs"], |
106 | 103 | ) |
| 104 | + ) |
107 | 105 |
|
108 | 106 | return advisories |
109 | 107 |
|
|
0 commit comments