# SPDX-FileCopyrightText: Huawei Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
# Create a diff of two JSON CVE statuses
import sys
import getopt
import re
from deepdiff import DeepDiff
from pprint import pprint
verbose_level = 0
def show_syntax_and_exit(code):
"""
Show the program syntax and exit with an errror
Arguments:
code: the error code to return
"""
print("Syntax: %s [-h] [-v] file1 file2" % __name__)
sys.exit(code)
def exit_error(code, message):
"""
Show the error message and exit with an errror
Arguments:
code: the error code to return
message: the message to show
"""
print("Error: %s" % message)
sys.exit(code)
def vprint(*args, **kwargs):
"""
Printing with verbosity levels
Arguments:
args: as for print
kwargs: as for print
"""
global verbose_level
if verbose_level:
print(*args, **kwargs)
def parse_args(argv):
"""
Parse the program arguments, put options in global variables
Arguments:
argv: program arguments
Returns:
Two mandatory file arguments
"""
global verbose_level
try:
opts, args = getopt.getopt(argv, "hv", ["help", "verbose"])
except getopt.GetoptError:
show_syntax_and_exit(1)
for opt, arg in opts:
if opt in ("-h", "--help"):
show_syntax_and_exit(0)
elif opt in ("-v", "--verbose"):
verbose_level = 1
else:
show_syntax_and_exit(1)
if len(args) < 2:
exit_error(1, "Need at least two files to compare")
return (args[0], args[1])
def validate_cve_json(data):
"""
Check correctness of the loaded JSON data
Arguments:
data: loaded data
Returns:
Bool: True if file is correct, False otherwise
Error message: if the file has errors
"""
if not "version" in data or data["version"] != "1":
return False, "Unrecognized format version number"
if not "package" in data:
return False, "Mandatory 'package' key not found"
for package in data["package"]:
keys_in_package = {"name", "layer", "version", "issue"}
if keys_in_package - package.keys():
return False, "Missing a mandatory key in package: %s" % (
keys_in_package - package.keys()
)
for issue in package["issue"]:
keys_in_issue = {"id", "scorev2", "scorev3", "vector", "status"}
if keys_in_issue - issue.keys():
return (
False,
"Missing mandatory keys %s in 'issue' for the package '%s'"
% (keys_in_issue - issue.keys(), package_name),
)
return True, ""
def get_name(value):
"""
Function used for sorting, return the sorting key to use
Argument:
value: raw value
Return:
Package name to use for sorting
"""
return value["name"]
def load_cve_json(filename):
"""
Load the JSON file, return the resulting dictionary
Arguments:
filename: the file to open
Returns:
Parsed file as a dictionary
"""
import json
out = {}
try:
with open(filename, "r") as f:
out = json.load(f)
except FileNotFoundError:
exit_error(1, "Input file (%s) not found" % (filename))
except json.decoder.JSONDecodeError as error:
exit_error(1, "Malformed JSON file: %s" % str(error))
# Validate file
validated, error = validate_cve_json(out)
if not validated:
exit_error(1, error)
out_sorted = sorted(out["package"], key=get_name)
return out_sorted
def calculate_diff(data1, data2):
"""
Calculate a deep diff between two JSON data sets
Arguments:
data1: source data to compare (current)
data2: destination data to compare (upstream)
Returns:
Diff in the DeepDiff tree format
"""
ddiff = DeepDiff(data1, data2, ignore_order=True, view="tree")
return ddiff
def calculate_package_diff(diff):
"""
Parse a difference in packages from the deep diff
Arguments:
diff: the deep diff, tree format
Returns:
removed set: added packages list
added_set: removed packages list
"""
removed_set = []
added_set = []
vprint("Package status:")
if "iterable_item_added" in diff.keys():
for p in diff["iterable_item_added"]:
# Only p.t2 exists
# Assure only "root[XXXX]" items
if not re.search("^root\[[0-9]+\]$", p.path()):
# If we have a new issue
# if re.search("^root\[[0-9]+\]\['issue'\]\[[0-9]+\]$", p):
# print("New issue")
continue
vprint("Added package: %s %s" % (p.t2["name"], p.t2["version"]))
added_set.append(p.t2)
if "iterable_item_removed" in diff.keys():
for p in diff["iterable_item_removed"]:
# Only p.t1 exists
# Assure only "root[XXXX]" items
if not re.search("^root\[[0-9]+\]$", p.path()):
# print("removed: Not found in %s" % (p.t1))
continue
vprint("Removed package: %s %s" % (p.t1["name"], p.t1["version"]))
removed_set.append(p.t1)
if "values_changed" in diff.keys():
# print("Some values changed")
for p in diff["values_changed"]:
# Filter out name changes, this is addition/removal - removal of t1
if re.search("^root\[[0-9]+\]\['name'\]$", p.path()):
vprint("Removed package: %s %s" % (p.up.t1["name"], p.up.t1["version"]))
vprint("Added package: %s %s" % (p.up.t2["name"], p.up.t2["version"]))
removed_set.append(p.up.t1)
added_set.append(p.up.t2)
# TODO: Handle the special case of linux
else:
# If a high level property root[XXX]['somename'] of the item that was changed
if re.search("^root\[[0-9]+\]\['[a-z,A-Z,0-9]+'\]$", p.path()):
# If the parent has been removed, this item is to be ignored
# We take it into account only if paren't hasn't been removed
if not p.up.t1 in removed_set:
vprint("Found changed items for: %s" % (p.up.t1["name"]))
# product table varies when the high level has changed
# Format: [root][XX]['products'][XX]['product']
elif re.search(
"^root\[[0-9]+\]\['products'\]\[[0-9]+\]\['product'\]$", p.path()
):
if p.up.up.up.t1 not in removed_set:
vprint(
"Change of product names for package: %s %s"
% (p.up.up.up.t1["name"], p.up.up.up.t2["name"])
)
# TODO: product table varies when the high level has changed
# TODO: issue table version when the high level has changed
# else:
# print(p)
vprint("")
return (removed_set, added_set)
def calculate_cve_diff(diff):
"""
Parse a difference in CVEs from the deep diff
Arguments:
diff: the deep diff, tree format
Returns:
removed_cves: removed CVEs
added_cves: added CVEs
"""
added_cves = []
removed_cves = []
vprint("CVE status:")
# Look for added/removed CVEs
if "values_changed" in diff.keys():
# Differences like <root[42]['issue'][2562]['status'] t1:'Patched', t2:'Unpatched'>
for p in diff["values_changed"]:
if re.search("^root\[[0-9]+\]\['issue'\]\[[0-9]+\]\['status'\]$", p.path()):
if (p.t1 == "Patched" or p.t1 == "Ignored") and p.t2 == "Unpatched":
# New CVE
vprint(
"Unpatched CVE: %s (%s %s)"
% (
p.up.t2["id"],
p.up.up.up.t2["name"],
p.up.up.up.t2["version"],
)
)
added_cves.append(p.up.t2["id"])
elif p.t1 == "Unpatched" and (p.t2 == "Patched" or p.t2 == "Ignored"):
# Fixed CVE
vprint(
"Fixed CVE: %s (%s %s)"
% (
p.up.t2["id"],
p.up.up.up.t2["name"],
p.up.up.up.t2["version"],
)
)
removed_cves.append(p.up.t2["id"])
else:
vprint(
"Unknown status of CVE: %s (%s %s)"
% (
p.up.t2["id"],
p.up.up.up.t2["name"],
p.up.up.up.t2["version"],
)
)
# Other changes in the issue table
elif re.search("^root\[[0-9]+\]\['issue'\].*$", p.path()):
# The whole issue element has changed
if re.search("^root\[[0-9]+\]\['issue'\]\[[0-9]+\]$", p.path()):
if p.up.up.t1["name"] == p.up.up.t2["name"]:
vprint(
"Changed CVE for package %s %s"
% (p.up.up.t2["name"], p.up.up.t2["version"])
)
else:
# Some fields of the issue changed
if re.search(
"^root\[[0-9]+\]\['issue'\]\[[0-9]+\]\['[a-z,A-Z]+'\]$",
p.path(),
):
# If package names are different, this is a spurious diff
if p.up.up.up.t1["name"] == p.up.up.up.t2["name"]:
vprint(
"Changed CVE for package %s %s"
% (p.up.up.up.t2["name"], p.up.up.up.t2["version"])
)
vprint("")
return (removed_cves, added_cves)
def main(argv):
file1, file2 = parse_args(argv)
data1 = load_cve_json(file1)
data2 = load_cve_json(file2)
diff = calculate_diff(data1, data2)
removed_set, added_set = calculate_package_diff(diff)
removed_cves, added_cves = calculate_cve_diff(diff)
print("Summary:")
print(
"Package report: the new version adds %s packages and removes %d packages"
% (len(added_set), len(removed_set))
)
print(
"CVE report: the new version removes %s CVEs and adds %s CVEs"
% (len(removed_cves), len(added_cves))
)
if __name__ == "__main__":
main(sys.argv[1:])