feat: implement IriusRisk CSV threat parser

Authored by T. Walker - DefectDojo

Author: skywalke34

dojo/tools/iriusrisk/parser.py

@@ -1,3 +1,17 @@
+import csv
+import hashlib
+import io
+
+from dojo.models import Finding
+
+SEVERITY_MAPPING = {
+    "Very low": "Info",
+    "Low": "Low",
+    "Medium": "Medium",
+    "High": "High",
+}
+
+
 class IriusriskParser:
 
     def get_scan_types(self):
@@ -10,4 +24,63 @@ def get_description_for_scan_types(self, scan_type):
         return "Import IriusRisk threat model CSV exports."
 
     def get_findings(self, filename, test):
-        return []
+        content = filename.read()
+        if isinstance(content, bytes):
+            content = content.decode("utf-8")
+        reader = csv.DictReader(io.StringIO(content), delimiter=",", quotechar='"')
+        findings = []
+        for row in reader:
+            component = (row.get("Component") or "").strip()
+            use_case = (row.get("Use case") or "").strip()
+            source = (row.get("Source") or "").strip()
+            threat = (row.get("Threat") or "").strip()
+            risk_response = (row.get("Risk Response") or "").strip()
+            inherent_risk = (row.get("Inherent Risk") or "").strip()
+            current_risk = (row.get("Current Risk") or "").strip()
+            countermeasure_progress = (row.get("Countermeasure progress") or "").strip()
+            weakness_tests = (row.get("Weakness tests") or "").strip()
+            countermeasure_tests = (row.get("Countermeasure tests") or "").strip()
+            projected_risk = (row.get("Projected Risk") or "").strip()
+            owner = (row.get("Owner") or "").strip()
+
+            # Title: truncate to 150 chars with ellipsis if needed
+            title = threat[:147] + "..." if len(threat) > 150 else threat
+
+            severity = SEVERITY_MAPPING.get(current_risk, "Info")
+
+            # Build description with all available fields
+            description_parts = [
+                f"**Threat:** {threat}",
+                f"**Component:** {component}",
+                f"**Use Case:** {use_case}",
+                f"**Source:** {source}",
+                f"**Inherent Risk:** {inherent_risk}",
+                f"**Current Risk:** {current_risk}",
+                f"**Projected Risk:** {projected_risk}",
+                f"**Countermeasure Progress:** {countermeasure_progress}",
+                f"**Weakness Tests:** {weakness_tests}",
+                f"**Countermeasure Tests:** {countermeasure_tests}",
+            ]
+            if owner:
+                description_parts.append(f"**Owner:** {owner}")
+            description = "\n".join(description_parts)
+
+            # Unique ID for deduplication across reimports
+            unique_id = hashlib.sha256(
+                f"{component}|{threat}|{risk_response}".encode(),
+            ).hexdigest()
+
+            finding = Finding(
+                test=test,
+                title=title,
+                severity=severity,
+                description=description,
+                mitigation=risk_response,
+                component_name=component,
+                active=current_risk != "Very low",
+                static_finding=True,
+                dynamic_finding=False,
+                unique_id_from_tool=unique_id,
+            )
+            findings.append(finding)
+        return findings