make sonarqube hotspots sync work (#13206)

* sonarqube hotspots sync implementation

* params fix

* ruff fixes

* ruff fixes

Author: dshafranskiy-r7

dojo/tools/api_sonarqube/api_client.py

@@ -334,6 +334,31 @@ def get_hotspot_rule(self, rule_id):
             self.rules_cache.update({rule_id: rule})
         return rule
 
+    def get_hotspot(self, rule_id):
+        """
+        Get detailed information about a hotspot
+        :param rule_id:
+        :return:
+        """
+        rule = self.rules_cache.get(rule_id)
+        if not rule:
+            response = self.session.get(
+                url=f"{self.sonar_api_url}/hotspots/search",
+                params={"hotspots": rule_id},
+                headers=self.default_headers,
+                timeout=settings.REQUESTS_TIMEOUT,
+            )
+            if not response.ok:
+                msg = (
+                    f"Unable to get the hotspot rule {rule_id} "
+                    f"due to {response.status_code} - {response.content}"
+                )
+                raise Exception(msg)
+
+            rule = response.json()["hotspots"][0]
+            self.rules_cache.update({rule_id: rule})
+        return rule
+
     def transition_issue(self, issue_key, transition):
         """
         Do workflow transition on an issue. Requires authentication and Browse permission on project.
@@ -375,6 +400,46 @@ def transition_issue(self, issue_key, transition):
             )
             raise Exception(msg)
 
+    def transition_hotspot(self, issue_key, status, resolution=None):
+        """
+        Do workflow transition on an issue. Requires authentication and Browse permission on project.
+        The transitions 'wontfix' and 'falsepositive' require the permission 'Administer Issues'.
+        The transitions involving security hotspots (except 'requestreview') require
+        the permission 'Administer Security Hotspot'.
+
+        Possible resolution values:
+        - FIXED
+        - SAFE
+        - ACKNOWLEDGED
+
+        Possible status values:
+        - TO_REVIEW
+        - REVIEWED
+
+        :param issue_key:
+        :param status:
+        :param resolution:
+        :return:
+        """
+        data = {"hotspot": issue_key, "status": status}
+
+        if resolution:
+            data["resolution"] = resolution
+
+        response = self.session.post(
+            url=f"{self.sonar_api_url}/hotspots/change_status",
+            data=data,
+            headers=self.default_headers,
+            timeout=settings.REQUESTS_TIMEOUT,
+        )
+
+        if not response.ok:
+            msg = (
+                f"Unable to change status {status} / resolution {resolution} the issue {issue_key} "
+                f'due to {response.status_code} - {response.content.decode("utf-8")}'
+            )
+            raise Exception(msg)
+
     def add_comment(self, issue_key, text):
         """
         Add a comment.

dojo/tools/api_sonarqube/updater.py

@@ -52,6 +52,45 @@ class SonarQubeApiUpdater:
         },
     ]
 
+    MAPPING_SONARQUBE_HOTSPOT_STATUS_TRANSITION = [
+        {
+            "from": ["TO_REVIEW"],
+            "to": "RESOLVED / FALSE-POSITIVE",
+            "transition": "REVIEWED",
+            "resolution": "SAFE",
+        },
+        {
+            "from": ["TO_REVIEW"],
+            "to": "RESOLVED / FIXED",
+            "transition": "REVIEWED",
+            "resolution": "FIXED",
+        },
+                {
+            "from": ["TO_REVIEW"],
+            "to": "RESOLVED / WONTFIX",
+            "transition": "REVIEWED",
+            "resolution": "ACKNOWLEDGED",
+        },
+        {
+            "from": ["REVIEWED"],
+            "to": "OPEN",
+            "transition": "TO_REVIEW",
+            "resolution": None,
+        },
+        {
+            "from": ["REVIEWED"],
+            "to": "REOPENED",
+            "transition": "TO_REVIEW",
+            "resolution": None,
+        },
+        {
+            "from": ["REVIEWED"],
+            "to": "CONFIRMED",
+            "transition": "TO_REVIEW",
+            "resolution": None,
+        },
+    ]
+
     @staticmethod
     def get_sonarqube_status_for(finding):
         target_status = None
@@ -66,16 +105,22 @@ def get_sonarqube_status_for(finding):
         return target_status
 
     def get_sonarqube_required_transitions_for(
-        self, current_status, target_status,
-    ):
+        self, current_status, target_status, is_hotspot):
         # If current and target is the same... do nothing
         if current_status == target_status:
             return None
 
+        # Select the appropriate mapping based on issue type
+        mapping = (
+            self.MAPPING_SONARQUBE_HOTSPOT_STATUS_TRANSITION
+            if is_hotspot
+            else self.MAPPING_SONARQUBE_STATUS_TRANSITION
+        )
+
         # Check if there is at least one transition from current_status...
         if not [
             x
-            for x in self.MAPPING_SONARQUBE_STATUS_TRANSITION
+            for x in mapping
             if current_status in x.get("from")
         ]:
             return None
@@ -84,34 +129,46 @@ def get_sonarqube_required_transitions_for(
         # can transition to target_status
         transitions = [
             x
-            for x in self.MAPPING_SONARQUBE_STATUS_TRANSITION
+            for x in mapping
             if target_status == x.get("to")
         ]
         if transitions:
             for transition in transitions:
                 # There is a direct transition from current status...
                 if current_status in transition.get("from"):
                     t = transition.get("transition")
+                    if is_hotspot:
+                        return [{"status": t, "resolution": transition.get("resolution")}] if t else None
                     return [t] if t else None
 
-            # We have the last transition to get to our target status but there
-            # is no direct transition
-            transitions_result = deque()
-            transitions_result.appendleft(transitions[0].get("transition"))
+            # Handle complex transitions for regular issues
+            if not is_hotspot:
+                # We have the last transition to get to our target status but there
+                # is no direct transition
+                transitions_result = deque()
+                transitions_result.appendleft(transitions[0].get("transition"))
 
-            # Find out previous transitions that would finish in any FROM of a
-            # previous to use as target
-            for transition in transitions:
-                for t_from in transition.get("from"):
-                    possible_transition = (
-                        self.get_sonarqube_required_transitions_for(
-                            current_status, t_from,
+                # Find out previous transitions that would finish in any FROM of a
+                # previous to use as target
+                for transition in transitions:
+                    for t_from in transition.get("from"):
+                        possible_transition = (
+                            self.get_sonarqube_required_transitions_for(
+                                current_status, t_from, is_hotspot,
+                            )
                         )
-                    )
-                    if possible_transition:
-                        transitions_result.extendleft(possible_transition)
-                        return list(transitions_result)
-            return None
+                        if possible_transition:
+                            transitions_result.extendleft(possible_transition)
+                            return list(transitions_result)
+            else:
+                # SQ code is too complicated for ISSUES, there is no such thing for HOTSPOTS,
+                # there are only 2 states: TO_REVIEW and REVIEWED
+                transitions_result = deque()
+                transitions_result.appendleft(
+                    {"status": transitions[0].get("transition"),
+                    "resolution": transitions[0].get("resolution")},
+                )
+                return list(transitions_result)
         return None
 
     def update_sonarqube_finding(self, finding):
@@ -128,44 +185,56 @@ def update_sonarqube_finding(self, finding):
         # during import
 
         target_status = self.get_sonarqube_status_for(finding)
+        is_hotspot = sonarqube_issue.type == "SECURITY_HOTSPOT"
 
-        issue = client.get_issue(sonarqube_issue.key)
-        if (
-            issue
-        ):  # Issue could have disappeared in SQ because a previous scan has resolved the issue as fixed
-            if issue.get("resolution"):
-                current_status = "{} / {}".format(
-                    issue.get("status"), issue.get("resolution"),
-                )
-            else:
-                current_status = issue.get("status")
+        issue = client.get_hotspot(sonarqube_issue.key) if is_hotspot else client.get_issue(sonarqube_issue.key)
+
+        # Issue does not exist (could have disappeared in SQ because a previous scan resolved it)
+        if not issue:
+            return
+
+        if is_hotspot:
+            current_status = issue.get("status")
+        elif issue.get("resolution"):
+            current_status = "{} / {}".format(issue.get("status"), issue.get("resolution"))
+        else:
+            current_status = issue.get("status")
 
+        # Get required transitions
+        transitions = self.get_sonarqube_required_transitions_for(
+            current_status, target_status, is_hotspot=is_hotspot,
+        )
+
+        if not transitions:
             logger.debug(
-                f"--> SQ Current status: {current_status}. Current target status: {target_status}",
+                f"There are no transitions between {current_status} and {target_status} for finding '{finding}' in SonarQube",
             )
+            return
 
-            transitions = self.get_sonarqube_required_transitions_for(
-                current_status, target_status,
+        logger.debug(
+                f"Updating finding '{finding}' transition {current_status} -> {target_status} in SonarQube",
             )
-            if transitions:
-                logger.info(
-                    f"Updating finding '{finding}' in SonarQube",
-                )
 
-                for transition in transitions:
-                    client.transition_issue(sonarqube_issue.key, transition)
-
-                # Track Defect Dojo has updated the SonarQube issue
-                Sonarqube_Issue_Transition.objects.create(
-                    sonarqube_issue=finding.sonarqube_issue,
-                    # not sure if this is needed, but looks like the original author decided to send display status
-                    # to sonarqube we changed Accepted into Risk Accepted, but we change it back to be sure we don't
-                    # break the integration
-                    finding_status=finding.status().replace(
-                        "Risk Accepted", "Accepted",
-                    )
-                    if finding.status()
-                    else finding.status(),
-                    sonarqube_status=current_status,
-                    transitions=",".join(transitions),
-                )
+        # Apply transitions
+        for transition in transitions:
+            if is_hotspot:
+                client.transition_hotspot(sonarqube_issue.key,
+                            status=transition["status"],
+                            resolution=transition["resolution"])
+            else:
+                client.transition_issue(sonarqube_issue.key, transition)
+
+        # Track that Defect Dojo has updated the SonarQube issue
+        Sonarqube_Issue_Transition.objects.create(
+            sonarqube_issue=finding.sonarqube_issue,
+            # not sure if this is needed, but looks like the original author decided to send display status
+            # to sonarqube we changed Accepted into Risk Accepted, but we change it back to be sure we don't
+            # break the integration
+            finding_status=finding.status().replace(
+            "Risk Accepted", "Accepted",
+            )
+            if finding.status()
+            else finding.status(),
+            sonarqube_status=current_status,
+            transitions=",".join(transition["status"] if is_hotspot else transition for transition in transitions),
+        )