Skip to content

Commit 0ff017f

Browse files
manuel-sommermanuel-sommer
authored
🎉 implement new threatmapper file format #13639 (#13655)
* 🎉 implement new threatmapper file format * update
1 parent 99a1d7e commit 0ff017f

File tree

10 files changed

+292
-67
lines changed

10 files changed

+292
-67
lines changed

‎dojo/tools/deepfence_threatmapper/compliance.py‎

Lines changed: 70 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,13 @@
33

44
class DeepfenceThreatmapperCompliance:
55
def get_findings(self, row, headers, test):
6-
description = ""
6+
if "compliance_check_type" in headers and "test_number" in headers:
7+
return self._parse_old_format(row, headers, test)
8+
if "Compliance Standard" in headers and "Control ID" in headers:
9+
return self._parse_new_format(row, headers, test)
10+
return None
11+
12+
def _parse_old_format(self, row, headers, test):
713
compliance_check_type = row[headers["compliance_check_type"]]
814
count = row[headers["count"]]
915
doc_id = row[headers["doc_id"]]
@@ -18,34 +24,76 @@ def get_findings(self, row, headers, test):
1824
test_desc = row[headers["test_desc"]]
1925
test_info = row[headers["test_info"]]
2026
test_number = row[headers["test_number"]]
21-
description += "**compliance_check_type:** " + str(compliance_check_type) + "\n"
22-
description += "**host_name:** " + str(host_name) + "\n"
23-
description += "**cloud_account_id:** " + str(cloud_account_id) + "\n"
24-
description += "**masked:** " + str(masked) + "\n"
25-
description += "**node_id:** " + str(node_id) + "\n"
26-
description += "**node_name:** " + str(node_name) + "\n"
27-
description += "**node_type:** " + str(node_type) + "\n"
28-
description += "**status:** " + str(status) + "\n"
29-
description += "**test_category:** " + str(test_category) + "\n"
30-
description += "**test_desc:** " + str(test_desc) + "\n"
31-
description += "**test_info:** " + str(test_info) + "\n"
32-
description += "**test_number:** " + str(test_number) + "\n"
33-
description += "**count:** " + str(count) + "\n"
34-
description += "**doc_id:** " + str(doc_id) + "\n"
27+
28+
description = (
29+
f"**Compliance Check Type:** {compliance_check_type}\n"
30+
f"**Host Name:** {host_name}\n"
31+
f"**Cloud Account ID:** {cloud_account_id}\n"
32+
f"**Masked:** {masked}\n"
33+
f"**Node ID:** {node_id}\n"
34+
f"**Node Name:** {node_name}\n"
35+
f"**Node Type:** {node_type}\n"
36+
f"**Status:** {status}\n"
37+
f"**Test Category:** {test_category}\n"
38+
f"**Test Description:** {test_desc}\n"
39+
f"**Test Info:** {test_info}\n"
40+
f"**Test Number:** {test_number}\n"
41+
f"**Count:** {count}\n"
42+
f"**Doc ID:** {doc_id}\n"
43+
)
44+
45+
return Finding(
46+
title=f"Threatmapper_Compliance_Report-{test_number}",
47+
description=description,
48+
severity=self.compliance_severity(status),
49+
static_finding=False,
50+
dynamic_finding=True,
51+
test=test,
52+
)
53+
54+
def _parse_new_format(self, row, headers, test):
55+
compliance_standard = row[headers["Compliance Standard"]]
56+
status = row[headers["Status"]]
57+
category = row[headers["Category"]]
58+
description_text = row[headers["Description"]]
59+
info = row[headers["Info"]]
60+
control_id = row[headers["Control ID"]]
61+
node_name = row[headers["Node Name"]]
62+
node_type = row[headers["Node Type"]]
63+
remediation = row[headers["Remediation"]]
64+
masked = row[headers["Masked"]]
65+
66+
description = (
67+
f"**Compliance Standard:** {compliance_standard}\n"
68+
f"**Status:** {status}\n"
69+
f"**Category:** {category}\n"
70+
f"**Description:** {description_text}\n"
71+
f"**Info:** {info}\n"
72+
f"**Control ID:** {control_id}\n"
73+
f"**Node Name:** {node_name}\n"
74+
f"**Node Type:** {node_type}\n"
75+
f"**Remediation:** {remediation}\n"
76+
f"**Masked:** {masked}\n"
77+
)
78+
3579
return Finding(
36-
title="Threatmapper_Compliance_Report-" + test_number,
80+
title=f"Threatmapper_Compliance_Report-{control_id}",
3781
description=description,
3882
severity=self.compliance_severity(status),
3983
static_finding=False,
4084
dynamic_finding=True,
85+
mitigation=remediation,
4186
test=test,
4287
)
4388

4489
def compliance_severity(self, severity_input):
90+
if severity_input is None:
91+
return "Info"
92+
severity_input = severity_input.lower()
4593
if severity_input in {"pass", "info"}:
46-
output = "Info"
47-
elif severity_input == "warn":
48-
output = "Medium"
49-
else:
50-
output = "Info"
51-
return output
94+
return "Info"
95+
if severity_input == "warn":
96+
return "Medium"
97+
if severity_input == "fail":
98+
return "High"
99+
return "Info"

‎dojo/tools/deepfence_threatmapper/malware.py‎

Lines changed: 49 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,13 @@
33

44
class DeepfenceThreatmapperMalware:
55
def get_findings(self, row, headers, test):
6-
description = ""
6+
if "Rule Name" in headers and "Class" in headers:
7+
return self._parse_old_format(row, headers, test)
8+
if "Rule Name" in headers and "Node Type" in headers:
9+
return self._parse_new_format(row, headers, test)
10+
return None
11+
12+
def _parse_old_format(self, row, headers, test):
713
Rule_Name = row[headers["Rule Name"]]
814
Class = row[headers["Class"]]
915
File_Name = row[headers["File Name"]]
@@ -13,14 +19,48 @@ def get_findings(self, row, headers, test):
1319
NodeType = row[headers["NodeType"]]
1420
Container_Name = row[headers["Container Name"]]
1521
Kubernetes_Cluster_Name = row[headers["Kubernetes Cluster Name"]]
16-
description += "**Summary:** " + str(Summary) + "\n"
17-
description += "**Rule Name:** " + str(Rule_Name) + "\n"
18-
description += "**Class:** " + str(Class) + "\n"
19-
description += "**File Name:** " + str(File_Name) + "\n"
20-
description += "**Node Name:** " + str(Node_Name) + "\n"
21-
description += "**NodeType:** " + str(NodeType) + "\n"
22-
description += "**Container Name:** " + str(Container_Name) + "\n"
23-
description += "**Kubernetes Cluster Name:** " + str(Kubernetes_Cluster_Name) + "\n"
22+
23+
description = (
24+
f"**Summary:** {Summary}\n"
25+
f"**Rule Name:** {Rule_Name}\n"
26+
f"**Class:** {Class}\n"
27+
f"**File Name:** {File_Name}\n"
28+
f"**Node Name:** {Node_Name}\n"
29+
f"**NodeType:** {NodeType}\n"
30+
f"**Container Name:** {Container_Name}\n"
31+
f"**Kubernetes Cluster Name:** {Kubernetes_Cluster_Name}\n"
32+
)
33+
34+
return Finding(
35+
title=Rule_Name,
36+
description=description,
37+
file_path=File_Name,
38+
severity=self.severity(Severity),
39+
static_finding=False,
40+
dynamic_finding=True,
41+
test=test,
42+
)
43+
44+
def _parse_new_format(self, row, headers, test):
45+
Rule_Name = row[headers["Rule Name"]]
46+
File_Name = row[headers["File Name"]]
47+
Summary = row[headers["Summary"]]
48+
Severity = row[headers["Severity"]]
49+
Node_Name = row[headers["Node Name"]]
50+
Node_Type = row[headers["Node Type"]]
51+
Kubernetes_Cluster_Name = row[headers["Kubernetes Cluster Name"]]
52+
Masked = row[headers["Masked"]]
53+
54+
description = (
55+
f"**Summary:** {Summary}\n"
56+
f"**Rule Name:** {Rule_Name}\n"
57+
f"**File Name:** {File_Name}\n"
58+
f"**Node Name:** {Node_Name}\n"
59+
f"**Node Type:** {Node_Type}\n"
60+
f"**Kubernetes Cluster Name:** {Kubernetes_Cluster_Name}\n"
61+
f"**Masked:** {Masked}\n"
62+
)
63+
2464
return Finding(
2565
title=Rule_Name,
2666
description=description,

‎dojo/tools/deepfence_threatmapper/parser.py‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,14 +27,23 @@ def get_findings(self, filename, test):
2727
first = False
2828
for i in range(len(row)):
2929
headers[row[i]] = i
30-
elif headers.get("Rule Name") is not None and headers.get("Class") is not None:
30+
elif (
31+
("Rule Name" in headers and "Class" in headers) or
32+
("Rule Name" in headers and "Node Type" in headers)
33+
):
3134
findings.append(DeepfenceThreatmapperMalware().get_findings(row, headers, test))
3235
elif headers.get("Filename") is not None and headers.get("Content") is not None:
3336
value = DeepfenceThreatmapperSecret().get_findings(row, headers, test)
3437
if value is not None:
3538
findings.append(value)
36-
elif headers.get("@timestamp") is not None and headers.get("cve_attack_vector") is not None:
39+
elif (
40+
("cve_id" in headers and "cve_attack_vector" in headers) or
41+
("CVE ID" in headers and "Attack Vector" in headers)
42+
):
3743
findings.append(DeepfenceThreatmapperVulnerability().get_findings(row, headers, test))
38-
elif headers.get("@timestamp") is not None and headers.get("compliance_check_type") is not None:
44+
elif (
45+
("compliance_check_type" in headers and "test_number" in headers) or
46+
("Compliance Standard" in headers and "Control ID" in headers)
47+
):
3948
findings.append(DeepfenceThreatmapperCompliance().get_findings(row, headers, test))
4049
return findings

‎dojo/tools/deepfence_threatmapper/secret.py‎

Lines changed: 57 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,13 @@
33

44
class DeepfenceThreatmapperSecret:
55
def get_findings(self, row, headers, test):
6+
if "Name" in headers and "Signature" in headers:
7+
return self._parse_old_format(row, headers, test)
8+
if "Content Starting Index" in headers and "Masked" in headers:
9+
return self._parse_new_format(row, headers, test)
10+
return None
11+
12+
def _parse_old_format(self, row, headers, test):
613
description = ""
714
Filename = row[headers["Filename"]]
815
Content = row[headers["Content"]]
@@ -13,27 +20,57 @@ def get_findings(self, row, headers, test):
1320
Container_Name = row[headers["Container Name"]]
1421
Kubernetes_Cluster_Name = row[headers["Kubernetes Cluster Name"]]
1522
Signature = row[headers["Signature"]]
16-
description += "**Filename:** " + str(Filename) + "\n"
17-
description += "**Name:** " + str(Name) + "\n"
18-
description += "**Rule:** " + str(Rule) + "\n"
19-
description += "**Node Name:** " + str(Node_Name) + "\n"
20-
description += "**Container Name:** " + str(Container_Name) + "\n"
21-
description += "**Kubernetes Cluster Name:** " + str(Kubernetes_Cluster_Name) + "\n"
22-
description += "**Content:** " + str(Content) + "\n"
23-
description += "**Signature:** " + str(Signature) + "\n"
24-
if Name is not None and Severity is not None:
25-
finding = Finding(
26-
title=str(Name),
27-
description=description,
28-
file_path=Filename,
29-
severity=self.severity(Severity),
30-
static_finding=False,
31-
dynamic_finding=True,
32-
test=test,
23+
description += f"**Filename:** {Filename}\n"
24+
description += f"**Name:** {Name}\n"
25+
description += f"**Rule:** {Rule}\n"
26+
description += f"**Node Name:** {Node_Name}\n"
27+
description += f"**Container Name:** {Container_Name}\n"
28+
description += f"**Kubernetes Cluster Name:** {Kubernetes_Cluster_Name}\n"
29+
description += f"**Content:** {Content}\n"
30+
description += f"**Signature:** {Signature}\n"
31+
if Name and Severity:
32+
return Finding(
33+
title=str(Name),
34+
description=description,
35+
file_path=Filename,
36+
severity=self.severity(Severity),
37+
static_finding=False,
38+
dynamic_finding=True,
39+
test=test,
40+
)
41+
return None
42+
43+
def _parse_new_format(self, row, headers, test):
44+
description = ""
45+
Filename = row[headers["Filename"]]
46+
Content = row[headers["Content"]]
47+
Rule = row[headers["Rule"]]
48+
Severity = row[headers["Severity"]]
49+
Content_Starting_Index = row[headers["Content Starting Index"]]
50+
Node_Name = row[headers["Node Name"]]
51+
Node_Type = row[headers["Node Type"]]
52+
Kubernetes_Cluster_Name = row[headers["Kubernetes Cluster Name"]]
53+
Masked = row[headers["Masked"]]
54+
description += f"**Filename:** {Filename}\n"
55+
description += f"**Rule:** {Rule}\n"
56+
description += f"**Node Name:** {Node_Name}\n"
57+
description += f"**Node Type:** {Node_Type}\n"
58+
description += f"**Kubernetes Cluster Name:** {Kubernetes_Cluster_Name}\n"
59+
description += f"**Content:** {Content}\n"
60+
description += f"**Content Starting Index:** {Content_Starting_Index}\n"
61+
description += f"**Masked:** {Masked}\n"
62+
title = f"{Rule} in {Filename}" if Rule else "Secret Finding"
63+
if Severity:
64+
return Finding(
65+
title=title,
66+
description=description,
67+
file_path=Filename,
68+
severity=self.severity(Severity),
69+
static_finding=False,
70+
dynamic_finding=True,
71+
test=test,
3372
)
34-
else:
35-
finding = None
36-
return finding
73+
return None
3774

3875
def severity(self, severity_input):
3976
if severity_input is None:

0 commit comments

Comments
 (0)