Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix github parser issue 9582 #9583

Merged
merged 5 commits into from
Mar 4, 2024
Merged

Fix github parser issue 9582 #9583

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fb4e979
add unittest file
manuel-sommer Feb 19, 2024
d557976
add unittest
manuel-sommer Feb 19, 2024
6919f80
adapt parser
manuel-sommer Feb 19, 2024
d03e3dc
fix unittest
manuel-sommer Feb 19, 2024
9ff3ed3
flake8
manuel-sommer Feb 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
287 changes: 162 additions & 125 deletions dojo/tools/github_vulnerability/parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,137 +17,174 @@ def get_description_for_scan_types(self, scan_type):

def get_findings(self, filename, test):
data = json.load(filename)
if "data" not in data:
raise ValueError("Invalid report file, no 'data' node found")

vulnerabilityAlerts = self._search_vulnerability_alerts(data["data"])
if not vulnerabilityAlerts:
raise ValueError(
"Invalid report, no 'vulnerabilityAlerts' node found"
)

repository_url = None
if "repository" in data["data"]:
if "nameWithOwner" in data["data"]["repository"]:
repository_url = "https://github.com/{}".format(
data["data"]["repository"]["nameWithOwner"]
)
if "url" in data["data"]["repository"]:
repository_url = data["data"]["repository"]["url"]

dupes = dict()
for alert in vulnerabilityAlerts["nodes"]:
description = alert["securityVulnerability"]["advisory"][
"description"
]

if "number" in alert and repository_url is not None:
dependabot_url = (
repository_url
+ "/security/dependabot/{}".format(alert["number"])
)
description = (
"[{}]({})\n".format(dependabot_url, dependabot_url)
+ description
if "data" in data:
vulnerabilityAlerts = self._search_vulnerability_alerts(data["data"])
if not vulnerabilityAlerts:
raise ValueError(
"Invalid report, no 'vulnerabilityAlerts' node found"
)

finding = Finding(
title=alert["securityVulnerability"]["advisory"]["summary"],
test=test,
description=description,
severity=self._convert_security(
alert["securityVulnerability"].get("severity", "MODERATE")
),
static_finding=True,
dynamic_finding=False,
unique_id_from_tool=alert["id"],
)

if "vulnerableManifestPath" in alert:
finding.file_path = alert["vulnerableManifestPath"]

if "vulnerableRequirements" in alert and alert["vulnerableRequirements"].startswith("= "):
finding.component_version = alert["vulnerableRequirements"][2:]

if "createdAt" in alert:
finding.date = dateutil.parser.parse(alert["createdAt"])

if "state" in alert and (
"FIXED" == alert["state"] or "DISMISSED" == alert["state"]
):
finding.active = False
finding.is_mitigated = True

# if the package is present
if "package" in alert["securityVulnerability"]:
finding.component_name = alert["securityVulnerability"][
"package"
].get("name")

if "references" in alert["securityVulnerability"]["advisory"]:
finding.references = ""
for ref in alert["securityVulnerability"]["advisory"][
"references"
]:
finding.references += ref["url"] + "\r\n"

if "identifiers" in alert["securityVulnerability"]["advisory"]:
unsaved_vulnerability_ids = list()
for identifier in alert["securityVulnerability"]["advisory"][
"identifiers"
]:
if identifier.get("value"):
unsaved_vulnerability_ids.append(
identifier.get("value")
)
if unsaved_vulnerability_ids:
finding.unsaved_vulnerability_ids = (
unsaved_vulnerability_ids
repository_url = None
if "repository" in data["data"]:
if "nameWithOwner" in data["data"]["repository"]:
repository_url = "https://github.com/{}".format(
data["data"]["repository"]["nameWithOwner"]
)

if "cvss" in alert["securityVulnerability"]["advisory"]:
if (
"score"
in alert["securityVulnerability"]["advisory"]["cvss"]
if "url" in data["data"]["repository"]:
repository_url = data["data"]["repository"]["url"]
dupes = dict()
for alert in vulnerabilityAlerts["nodes"]:
description = alert["securityVulnerability"]["advisory"][
"description"
]
if "number" in alert and repository_url is not None:
dependabot_url = (
repository_url
+ "/security/dependabot/{}".format(alert["number"])
)
description = (
"[{}]({})\n".format(dependabot_url, dependabot_url)
+ description
)
finding = Finding(
title=alert["securityVulnerability"]["advisory"]["summary"],
test=test,
description=description,
severity=self._convert_security(
alert["securityVulnerability"].get("severity", "MODERATE")
),
static_finding=True,
dynamic_finding=False,
unique_id_from_tool=alert["id"],
)
if "vulnerableManifestPath" in alert:
finding.file_path = alert["vulnerableManifestPath"]
if "vulnerableRequirements" in alert and alert["vulnerableRequirements"].startswith("= "):
finding.component_version = alert["vulnerableRequirements"][2:]
if "createdAt" in alert:
finding.date = dateutil.parser.parse(alert["createdAt"])
if "state" in alert and (
"FIXED" == alert["state"] or "DISMISSED" == alert["state"]
):
score = alert["securityVulnerability"]["advisory"]["cvss"][
finding.active = False
finding.is_mitigated = True
# if the package is present
if "package" in alert["securityVulnerability"]:
finding.component_name = alert["securityVulnerability"][
"package"
].get("name")
if "references" in alert["securityVulnerability"]["advisory"]:
finding.references = ""
for ref in alert["securityVulnerability"]["advisory"][
"references"
]:
finding.references += ref["url"] + "\r\n"
if "identifiers" in alert["securityVulnerability"]["advisory"]:
unsaved_vulnerability_ids = list()
for identifier in alert["securityVulnerability"]["advisory"][
"identifiers"
]:
if identifier.get("value"):
unsaved_vulnerability_ids.append(
identifier.get("value")
)
if unsaved_vulnerability_ids:
finding.unsaved_vulnerability_ids = (
unsaved_vulnerability_ids
)
if "cvss" in alert["securityVulnerability"]["advisory"]:
if (
"score"
]
if score is not None:
finding.cvssv3_score = score
in alert["securityVulnerability"]["advisory"]["cvss"]
):
score = alert["securityVulnerability"]["advisory"]["cvss"][
"score"
]
if score is not None:
finding.cvssv3_score = score
if (
"vectorString"
in alert["securityVulnerability"]["advisory"]["cvss"]
):
cvss_vector_string = alert["securityVulnerability"][
"advisory"
]["cvss"]["vectorString"]
if cvss_vector_string is not None:
cvss_objects = cvss_parser.parse_cvss_from_text(
cvss_vector_string
)
if len(cvss_objects) > 0:
finding.cvssv3 = cvss_objects[0].clean_vector()
if (
"vectorString"
in alert["securityVulnerability"]["advisory"]["cvss"]
"cwes" in alert["securityVulnerability"]["advisory"]
and "nodes"
in alert["securityVulnerability"]["advisory"]["cwes"]
):
cvss_vector_string = alert["securityVulnerability"][
"advisory"
]["cvss"]["vectorString"]
if cvss_vector_string is not None:
cvss_objects = cvss_parser.parse_cvss_from_text(
cvss_vector_string
)
if len(cvss_objects) > 0:
finding.cvssv3 = cvss_objects[0].clean_vector()

if (
"cwes" in alert["securityVulnerability"]["advisory"]
and "nodes"
in alert["securityVulnerability"]["advisory"]["cwes"]
):
cwe_nodes = alert["securityVulnerability"]["advisory"]["cwes"][
"nodes"
]
if cwe_nodes and len(cwe_nodes) > 0:
finding.cwe = int(cwe_nodes[0].get("cweId")[4:])

dupe_key = finding.unique_id_from_tool
if dupe_key in dupes:
find = dupes[dupe_key]
find.nb_occurences += 1
else:
dupes[dupe_key] = finding

return list(dupes.values())
cwe_nodes = alert["securityVulnerability"]["advisory"]["cwes"][
"nodes"
]
if cwe_nodes and len(cwe_nodes) > 0:
finding.cwe = int(cwe_nodes[0].get("cweId")[4:])
dupe_key = finding.unique_id_from_tool
if dupe_key in dupes:
find = dupes[dupe_key]
find.nb_occurences += 1
else:
dupes[dupe_key] = finding
return list(dupes.values())
elif isinstance(data, list):
findings = list()
for vuln in data:
url = vuln["url"]
html_url = vuln["html_url"]
if vuln["state"] == "open":
active = True
else:
active = False
ruleid = vuln["rule"]["id"]
ruleseverity = vuln["rule"]["severity"]
ruledescription = vuln["rule"]["description"]
rulename = vuln["rule"]["name"]
ruletags = vuln["rule"]["tags"]
severity = vuln["rule"]["security_severity_level"]
most_recent_instanceref = vuln["most_recent_instance"]["ref"]
most_recent_instanceanalysis_key = vuln["most_recent_instance"]["analysis_key"]
most_recent_instanceenvironment = vuln["most_recent_instance"]["environment"]
most_recent_instancecategory = vuln["most_recent_instance"]["category"]
most_recent_instancestate = vuln["most_recent_instance"]["state"]
most_recent_instancecommit_sha = vuln["most_recent_instance"]["commit_sha"]
most_recent_instancemessage = vuln["most_recent_instance"]["message"]["text"]
location = vuln["most_recent_instance"]["location"]
instancesurl = vuln["instances_url"]
description = ruledescription + "\n"
description += "**url:** " + url + "\n"
description += "**html_url:** " + html_url + "\n"
description += "**ruleid:** " + ruleid + "\n"
description += "**ruleseverity:** " + ruleseverity + "\n"
description += "**ruledescription:** " + ruledescription + "\n"
description += "**rulename:** " + rulename + "\n"
description += "**ruletags:** " + str(ruletags) + "\n"
description += "**most_recent_instanceref:** " + most_recent_instanceref + "\n"
description += "**most_recent_instanceanalysis_key:** " + most_recent_instanceanalysis_key + "\n"
description += "**most_recent_instanceenvironment:** " + most_recent_instanceenvironment + "\n"
description += "**most_recent_instancecategory:** " + most_recent_instancecategory + "\n"
description += "**most_recent_instancestate:** " + most_recent_instancestate + "\n"
description += "**most_recent_instancecommit_sha:** " + most_recent_instancecommit_sha + "\n"
description += "**most_recent_instancemessage:** " + most_recent_instancemessage + "\n"
description += "**location:** " + str(location) + "\n"
description += "**instancesurl:** " + instancesurl + "\n"
uniqueid = ruleid + url + most_recent_instanceanalysis_key + str(location)
finding = Finding(
title=ruleid,
test=test,
description=description,
severity=severity.capitalize(),
active=active,
static_finding=True,
dynamic_finding=False,
unique_id_from_tool=uniqueid,
)
findings.append(finding)
return findings

def _search_vulnerability_alerts(self, data):
if isinstance(data, list):
Expand Down
Loading
Loading