Add new parser - Threat Composer

docs/content/en/integrations/parsers/file/threat_composer.md

@@ -0,0 +1,9 @@
+---
+title: "Threat Composer"
+toc_hide: true
+---
+### File Types
+This DefectDojo parser accepts JSON files from Threat Composer. The tool supports the [export](https://github.com/awslabs/threat-composer/tree/main?#features) of JSON report out of the browser local storage to a local file.
+
+### Sample Scan Data
+Sample scan data for testing purposes can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/threat_composer).

dojo/settings/.settings.dist.py.sha256sum

@@ -1 +1 @@
-71285f56a01869df55a802d79343f43c2e6a42ed52c4bb3591202e62b8569c64
+5b5b80d9559990d731f28be5d02e2cdeafe00070c83174008cceaeec74fe1813

dojo/settings/settings.dist.py

@@ -1278,6 +1278,7 @@ def saml2_attrib_map_format(dict):
     "Rapplex Scan": ["title", "endpoints", "severity"],
     "AppCheck Web Application Scanner": ["title", "severity"],
     "Legitify Scan": ["title", "endpoints", "severity"],
+    "ThreatComposer Scan": ["title", "description"],
 }
 
 # Override the hardcoded settings here via the env var
@@ -1501,6 +1502,7 @@ def saml2_attrib_map_format(dict):
     "Rapplex Scan": DEDUPE_ALGO_HASH_CODE,
     "AppCheck Web Application Scanner": DEDUPE_ALGO_HASH_CODE,
     "Legitify Scan": DEDUPE_ALGO_HASH_CODE,
+    "ThreatComposer Scan": DEDUPE_ALGO_UNIQUE_ID_FROM_TOOL_OR_HASH_CODE,
 }
 
 # Override the hardcoded settings here via the env var

dojo/tools/threat_composer/parser.py

@@ -0,0 +1,152 @@
+import json
+from collections import defaultdict
+from os import linesep
+
+from dojo.models import Finding
+
+
+class ThreatComposerParser:
+    """
+    Threat Composer JSON can be imported. See here for more info on this JSON format.
+    """
+
+    PRIORITY_VALUES = ["Low", "Medium", "High"]
+    STRIDE_VALUES = {
+        "S": "Spoofing",
+        "T": "Tampering",
+        "R": "Repudiation",
+        "I": "Information Disclosure",
+        "D": "Denial of Service",
+        "E": "Elevation of Privilege",
+    }
+
+    def get_scan_types(self):
+        return ["ThreatComposer Scan"]
+
+    def get_label_for_scan_types(self, scan_type):
+        return "ThreatComposer Scan"
+
+    def get_description_for_scan_types(self, scan_type):
+        return "ThreatComposer report file can be imported in JSON format."
+
+    def get_findings(self, file, test):
+        data = json.load(file)
+        findings = []
+
+        if "threats" not in data:
+            msg = "Invalid ThreatComposer data"
+            raise ValueError(msg)
+
+        if "assumptionLinks" in data:
+            assumptions = {assumption["id"]: assumption for assumption in data["assumptions"]}
+            assumption_mitigation_links = defaultdict(list)
+            assumption_threat_links = defaultdict(list)
+            for link in data["assumptionLinks"]:
+                linked_id = link["linkedId"]
+                assumption_id = link["assumptionId"]
+                assumption_type = link["type"]
+                if assumption_id in assumptions:
+                    if assumption_type == "Threat":
+                        assumption_threat_links[linked_id].append(assumptions[assumption_id])
+                    elif assumption_type == "Mitigation":
+                        assumption_mitigation_links[linked_id].append(assumptions[assumption_id])
+
+        if "mitigationLinks" in data:
+            mitigations = {
+                mitigation["id"]: {
+                    "mitigation": mitigation,
+                    "assumptions": assumption_mitigation_links[mitigation["id"]],
+                }
+                for mitigation in data["mitigations"]
+            }
+            mitigation_links = defaultdict(list)
+            for link in data["mitigationLinks"]:
+                linked_id = link["linkedId"]
+                mitigation_id = link["mitigationId"]
+                if mitigation_id in mitigations:
+                    mitigation_links[linked_id].append(mitigations[mitigation_id])
+
+        for threat in data["threats"]:
+
+            if "threatAction" in threat:
+                title = threat["threatAction"]
+                severity, impact, comments = self.parse_threat_metadata(threat["metadata"])
+                description = self.to_description_text(threat, comments, assumption_threat_links[threat["id"]])
+                mitigation = self.to_mitigation_text(mitigation_links[threat["id"]])
+                unique_id_from_tool = threat["id"]
+                vuln_id_from_tool = threat["numericId"]
+                tags = threat["tags"] if "tags" in threat else []
+
+                finding = Finding(
+                    title=title,
+                    description=description,
+                    severity=severity,
+                    vuln_id_from_tool=vuln_id_from_tool,
+                    unique_id_from_tool=unique_id_from_tool,
+                    mitigation=mitigation,
+                    impact=impact,
+                    tags=tags,
+                    static_finding=True,
+                    dynamic_finding=False,
+                )
+
+                match threat.get("status", "threatIdentified"):
+                    case "threatResolved":
+                        finding.active = False
+                        finding.is_mitigated = True
+                        finding.false_p = False
+                    case "threatResolvedNotUseful":
+                        finding.active = False
+                        finding.is_mitigated = True
+                        finding.false_p = True
+
+                findings.append(finding)
+
+        return findings
+
+    def to_mitigation_text(self, mitigations):
+        text = ""
+        for i, current in enumerate(mitigations):
+            mitigation = current["mitigation"]
+            assumption_links = current["assumptions"]
+            counti = i + 1
+            text += f"**Mitigation {counti} (ID: {mitigation['numericId']}, Status: {mitigation.get('status', 'Not defined')})**: {mitigation['content']}"
+
+            for item in mitigation["metadata"]:
+                if item["key"] == "Comments":
+                    text += f"\n*Comments*: {item['value'].replace(linesep, ' ')} "
+                    break
+
+            for j, assumption in enumerate(assumption_links):
+                countj = j + 1
+                text += f"\n- *Assumption {countj} (ID: {assumption['numericId']})*: {assumption['content'].replace(linesep, ' ')}"
+
+            text += "\n"
+
+        return text
+
+    def parse_threat_metadata(self, metadata):
+        severity = "Info"
+        impact = None
+        comments = None
+
+        for item in metadata:
+            if item["key"] == "Priority" and item["value"] in self.PRIORITY_VALUES:
+                severity = item["value"]
+            elif item["key"] == "STRIDE" and all(element in self.STRIDE_VALUES for element in item["value"]):
+                impact = ", ".join([self.STRIDE_VALUES[element] for element in item["value"]])
+            elif item["key"] == "Comments":
+                comments = item["value"]
+
+        return severity, impact, comments
+
+    def to_description_text(self, threat, comments, assumption_links):
+        text = f"**Threat**: {threat['statement']}"
+        if comments:
+            text += f"\n*Comments*: {comments}"
+
+        for i, assumption in enumerate(assumption_links):
+            counti = i + 1
+            text += f"\n- *Assumption {counti} (ID: {assumption['numericId']})*: {assumption['content'].replace(linesep, ' ')}"
+
+        return text

unittests/scans/threat_composer/threat_composer_broken_assumptions.json

@@ -0,0 +1,119 @@
+{
+    "schema": 1,
+    "applicationInfo": {
+      "name": "Threat composer",
+      "description": ""
+    },
+    "architecture": {
+      "image": "",
+      "description": ""
+    },
+    "dataflow": {
+      "image": "",
+      "description": ""
+    },
+    "assumptions": [
+      {
+        "id": "2d2a1ddf-5bb8-4a55-8f60-e195bc0b4b90",
+        "numericId": 7,
+        "content": "lorem ipsum",
+        "tags": [
+          "lorem ipsum"
+        ],
+        "metadata": [
+          {
+            "key": "Comments",
+            "value": "lorem ipsum"
+          }
+        ],
+        "displayOrder": 7
+      }
+    ],
+    "mitigations": [
+      {
+        "id": "bdef5b69-e690-4c9c-bfc1-960390779d3b",
+        "numericId": 21,
+        "content": "lorem ipsum",
+        "tags": [
+          "lorem ipsum"
+        ],
+        "metadata": [
+          {
+            "key": "Comments",
+            "value": "lorem ipsum"
+          }
+        ],
+        "displayOrder": 21
+      },
+      {
+        "id": "11fb1c71-42f0-4004-89a7-09d8bf6f8b11",
+        "numericId": 20,
+        "content": "lorem ipsum",
+        "metadata": [
+          {
+            "key": "Comments",
+            "value": "lorem ipsum"
+          }
+        ],
+        "displayOrder": 20
+      }
+    ],
+    "assumptionLinks": [
+      {
+        "linkedId": "46db1eb4-a451-4d05-afe1-c695491e2387",
+        "assumptionId": "d8edcf30-5c76-49f7-a408-20e071bbea1c",
+        "type": "Threat"
+      }
+    ],
+    "mitigationLinks": [
+      {
+        "linkedId": "46db1eb4-a451-4d05-afe1-c695491e2387",
+        "mitigationId": "11fb1c71-42f0-4004-89a7-09d8bf6f8b11"
+      },
+      {
+        "linkedId": "46db1eb4-a451-4d05-afe1-c695491e2387",
+        "mitigationId": "bdef5b69-e690-4c9c-bfc1-960390779d3b"
+      }
+    ],
+    "threats": [
+      {
+        "id": "46db1eb4-a451-4d05-afe1-c695491e2387",
+        "numericId": 23,
+        "statement": "A lorem ipsum lorem ipsum can lorem ipsum, which leads to lorem ipsum, negatively impacting lorem ipsum",
+        "threatSource": "lorem ipsum",
+        "prerequisites": "lorem ipsum",
+        "threatAction": "lorem ipsum",
+        "threatImpact": "lorem ipsum",
+        "impactedAssets": [
+          "lorem ipsum"
+        ],
+        "displayOrder": 23,
+        "metadata": [
+          {
+            "key": "Priority",
+            "value": "High"
+          },
+          {
+            "key": "STRIDE",
+            "value": [
+              "S",
+              "T",
+              "R",
+              "I",
+              "D",
+              "E"
+            ]
+          },
+          {
+            "key": "Comments",
+            "value": "lorem ipsum. lorem ipsum lorem ipsum"
+          }
+        ],
+        "tags": [
+          "CWE-156",
+          "CVE-45",
+          "lorem ipsum"
+        ]
+      }
+    ]
+  }