|
2 | 2 |
|
3 | 3 | import json |
4 | 4 | import os |
| 5 | +import typing |
5 | 6 | from textwrap import indent |
6 | 7 |
|
7 | 8 | import jsonschema |
8 | 9 |
|
| 10 | +from typings import osv |
| 11 | + |
9 | 12 | report_valid = False |
10 | 13 |
|
11 | 14 | with open('schema.json') as f: |
|
16 | 19 | total = 0 |
17 | 20 | passed = 0 |
18 | 21 |
|
| 22 | + |
| 23 | +def load_advisory_data(path_to_advisory: str) -> osv.Vulnerability: |
| 24 | + with open(path_to_advisory) as fi: |
| 25 | + return typing.cast(osv.Vulnerability, json.load(fi)) |
| 26 | + |
| 27 | + |
| 28 | +def has_database_specific_warnings(vuln: osv.Vulnerability) -> bool: |
| 29 | + if len(vuln.get('database_specific', {}).get('warnings', [])) > 0: |
| 30 | + return True |
| 31 | + |
| 32 | + for affected in vuln['affected']: |
| 33 | + if len(affected.get('database_specific', {}).get('warnings', [])) > 0: |
| 34 | + return True |
| 35 | + for ran in affected['ranges']: |
| 36 | + if len(ran.get('database_specific', {}).get('warnings', [])) > 0: |
| 37 | + return True |
| 38 | + return False |
| 39 | + |
| 40 | + |
19 | 41 | for dirpath, _, filenames in os.walk('advisories'): |
20 | 42 | for filename in filenames: |
21 | | - advisory = os.path.join(dirpath, filename) |
| 43 | + advisory_filepath = os.path.join(dirpath, filename) |
22 | 44 | total += 1 |
23 | 45 |
|
24 | 46 | try: |
25 | | - with open(advisory) as f: |
26 | | - validator.validate(json.load(f)) |
27 | | - if report_valid: |
28 | | - print(f'✅ {advisory} is valid') |
| 47 | + data = load_advisory_data(advisory_filepath) |
| 48 | + |
| 49 | + validator.validate(data) |
| 50 | + |
| 51 | + if has_database_specific_warnings(data): |
| 52 | + if 'CI' in os.environ: |
| 53 | + print( |
| 54 | + f'::warning file={advisory_filepath},line=1::has warnings that should be reviewed and patched if possible' |
| 55 | + ) |
| 56 | + print(f'⚠️ {advisory_filepath} is valid with warnings') |
| 57 | + elif report_valid: |
| 58 | + print(f'✅ {advisory_filepath} is valid') |
29 | 59 | passed += 1 |
30 | 60 | except (json.JSONDecodeError, jsonschema.ValidationError) as err: |
31 | | - print(f'❌ {advisory} is invalid') |
| 61 | + print(f'❌ {advisory_filepath} is invalid') |
32 | 62 | print(indent(str(err), prefix=' ')) |
33 | 63 |
|
34 | 64 | print(f'ℹ️ validated {total} advisories, with {total - passed} invalid') |
|
0 commit comments