Fixes #1314

Author: okorach

sonar/components.py

@@ -32,7 +32,7 @@
 import sonar.sqobject as sq
 import sonar.platform as pf
 
-from sonar import settings, tasks, measures, utilities
+from sonar import settings, tasks, measures, utilities, rules, issues
 
 import sonar.audit.problem as pb
 
@@ -137,6 +137,19 @@ def get_issues(self, filters: types.ApiParams = None) -> dict[str, object]:
         self.nbr_issues = len(issue_list)
         return issue_list
 
+    def count_third_party_issues(self, filters: types.ApiParams = None) -> dict[str, int]:
+        """Returns list of issues for a component, optionally on branches or/and PRs"""
+        from sonar.issues import component_filter
+
+        third_party_rules = rules.third_party(self.endpoint)
+        params = utilities.replace_keys(_ALT_COMPONENTS, component_filter(self.endpoint), self.search_params())
+        if filters is not None:
+            params.update(filters)
+        params["facets"] = "rules"
+        params["rules"] = [r.key for r in third_party_rules]
+        issues_count = {k: v for k, v in issues.count_by_rule(endpoint=self.endpoint, **params).items() if v > 0}
+        return issues_count
+
     def get_hotspots(self, filters: types.ApiParams = None) -> dict[str, object]:
         """Returns list of hotspots for a component, optionally on branches or/and PRs"""
         from sonar.hotspots import component_filter, search
@@ -161,8 +174,8 @@ def get_measures(self, metrics_list: types.KeyList) -> dict[str, any]:
 
     def get_measure(self, metric: str, fallback: int = None) -> any:
         """Returns a component measure"""
-        meas = self.get_measures(metric)
-        return meas[metric].value if metric in meas and meas[metric].value is not None else fallback
+        meas = self.get_measures([metric])
+        return meas[metric].value if metric in meas and meas[metric] and meas[metric].value is not None else fallback
 
     def loc(self) -> int:
         """Returns a component nbr of LOC"""

sonar/issues.py

@@ -818,14 +818,40 @@ def count(endpoint: pf.Platform, **kwargs) -> int:
     params = {} if not kwargs else kwargs.copy()
     params["ps"] = 1
     try:
-        log.info("Count params = %s", str(params))
+        log.debug("Count params = %s", str(params))
         nbr_issues = len(search(endpoint=endpoint, params=params))
     except TooManyIssuesError as e:
         nbr_issues = e.nbr_issues
     log.debug("Issue search %s would return %d issues", str(kwargs), nbr_issues)
     return nbr_issues
 
 
+def count_by_rule(endpoint: pf.Platform, **kwargs) -> dict[str, int]:
+    """Returns number of issues of a search"""
+    params = {} if not kwargs else kwargs.copy()
+    params["ps"] = 1
+    params["facets"] = "rules"
+    SLICE_SIZE = 50  # Search rules facets by bulks of 50
+    nbr_slices = 1
+    if "rules" in params:
+        nbr_slices = (len(params["rules"]) + SLICE_SIZE - 1) // SLICE_SIZE
+    rulecount = {}
+    for i in range(nbr_slices):
+        sliced_params = params.copy()
+        sliced_params["rules"] = ",".join(params["rules"][i * SLICE_SIZE : min((i + 1) * SLICE_SIZE - 1, len(params["rules"]))])
+        # log.debug("COUNT params = %s", str(sliced_params))
+        data = json.loads(endpoint.get(Issue.SEARCH_API, params=sliced_params).text)["facets"][0]["values"]
+        # log.debug("COUNT data results = %s", str(data))
+        for d in data:
+            if d["val"] not in params["rules"]:
+                continue
+            if d["val"] not in rulecount:
+                rulecount[d["val"]] = 0
+            rulecount[d["val"]] += d["count"]
+    log.debug("Rule counts = %s", util.json_dump(rulecount))
+    return rulecount
+
+
 def get_object(endpoint: pf.Platform, key: str, data: ApiPayload = None, from_export: bool = False) -> Issue:
     """Returns an issue from its key"""
     uu = sqobject.uuid(key, endpoint.url)

sonar/projects.py

@@ -42,7 +42,7 @@
 from sonar.util import types
 
 from sonar import exceptions, errcodes
-from sonar import sqobject, components, qualitygates, qualityprofiles, tasks, settings, webhooks, devops, syncer
+from sonar import sqobject, components, qualitygates, qualityprofiles, rules, tasks, settings, webhooks, devops, syncer
 import sonar.permissions.permissions as perms
 from sonar import pull_requests, branches
 import sonar.utilities as util
@@ -783,6 +783,20 @@ def get_issues(self, filters: dict[str, str] = None) -> dict[str, object]:
                 findings_list = {**findings_list, **comp.get_issues()}
         return findings_list
 
+    def count_third_party_issues(self, filters: dict[str, str] = None) -> dict[str, int]:
+        branches_or_prs = self.get_branches_and_prs(filters)
+        if branches_or_prs is None:
+            return super().count_third_party_issues(filters)
+        issue_counts = {}
+        for comp in branches_or_prs.values():
+            if not comp:
+                continue
+            for k, total in comp.count_third_party_issues(filters):
+                if k not in issue_counts:
+                    issue_counts[k] = 0
+                issue_counts[k] += total
+        return issue_counts
+
     def __sync_community(self, another_project: object, sync_settings: types.ConfigSettings) -> tuple[list[dict[str, str]], dict[str, int]]:
         """Syncs 2 projects findings on a community edition"""
         report, counters = [], {}
@@ -978,6 +992,8 @@ def export(self, export_settings: types.ConfigSettings, settings_list: dict[str,
                         "lastTaskWarnings": last_task.warnings(),
                         "taskHistory": [t._json for t in self.task_history()],
                     }
+                json_data["thirdPartyIssues"] = self.count_third_party_issues()
+                log.info("%s has %d 3rd party issues", str(self), sum(v for v in json_data["thirdPartyIssues"].values()))
 
             settings_dict = settings.get_bulk(endpoint=self.endpoint, component=self, settings_list=settings_list, include_not_set=False)
             # json_data.update({s.to_json() for s in settings_dict.values() if include_inherited or not s.inherited})

sonar/rules.py

@@ -41,6 +41,52 @@
 
 TYPES = ("BUG", "VULNERABILITY", "CODE_SMELL", "SECURITY_HOTSPOT")
 
+SONAR_REPOS = {
+    "abap",
+    "apex",
+    "azureresourcemanager",
+    "c",
+    "cloudformation",
+    "cobol",
+    "cpp",
+    "csharpsquid",
+    "roslyn.sonaranalyzer.security.cs",
+    "css",
+    "docker",
+    "flex",
+    "go",
+    "java",
+    "javabugs",
+    "javasecurity",
+    "javascript",
+    "jssecurity",
+    "jcl",
+    "kotlin",
+    "kubernetes",
+    "objc",
+    "php",
+    "phpsecurity",
+    "pli",
+    "plsql",
+    "python",
+    "pythonbugs",
+    "pythonsecurity",
+    "rpg",
+    "ruby",
+    "scala",
+    "secrets",
+    "swift",
+    "terraform",
+    "text",
+    "tsql",
+    "typescript",
+    "tssecurity",
+    "vb",
+    "vbnet",
+    "Web",
+    "xml",
+}
+
 
 class Rule(sq.SqObject):
     """
@@ -205,7 +251,9 @@ def count(endpoint: platform.Platform, **params) -> int:
 
 def get_list(endpoint: platform.Platform, **params) -> dict[str, Rule]:
     """Returns a list of rules corresponding to certain csearch filters"""
-    return search(endpoint, include_external="false", **params)
+    if len(_OBJECTS) < 100:
+        return search(endpoint, include_external="false", **params)
+    return _OBJECTS
 
 
 def get_object(endpoint: platform.Platform, key: str) -> Optional[Rule]:
@@ -249,6 +297,8 @@ def export(endpoint: platform.Platform, export_settings: types.ConfigSettings, k
         rule_list["extended"] = extended_rules
     if len(other_rules) > 0 and full:
         rule_list["standard"] = other_rules
+    if export_settings["MODE"] == "MIGRATION":
+        rule_list["thirdParty"] = {r.key: r.export() for r in third_party(endpoint=endpoint)}
     return rule_list
 
 
@@ -331,3 +381,8 @@ def convert_for_yaml(original_json: types.ObjectJsonRepr) -> types.ObjectJsonRep
         if category in original_json:
             new_json[category] = convert_rule_list_for_yaml(original_json[category])
     return new_json
+
+
+def third_party(endpoint: platform.Platform) -> list[Rule]:
+    """Returns the list of rules coming from 3rd party plugins"""
+    return [r for r in get_list(endpoint=endpoint).values() if r.repo and r.repo not in SONAR_REPOS and not r.repo.startswith("external_")]