Introduce Rescoring also for IPFindings

Signed-off-by: TuanAnh17N <[email protected]>

Author: TuanAnh17N

metadata.py

@@ -27,6 +27,7 @@
     odg.model.Datatype.DIKI_FINDING,
     odg.model.Datatype.OSID_FINDING,
     odg.model.Datatype.CRYPTO_FINDING,
+    odg.model.Datatype.IP_FINDING,
 )
 
 
@@ -531,6 +532,20 @@ def reuse_discovery_date_if_possible(
             # resource-/package-version, so we must re-use its discovery date
             return old_metadata.discovery_date
 
+    elif new_metadata.type == odg.model.Datatype.IP_FINDING:
+        if (
+            new_metadata.data.get('package_name') == old_metadata.data.get('package_name')
+            and new_metadata.data.get('license').get('name')
+                == old_metadata.data.get('license').get('name')
+            and sorted(new_metadata.data.get('labels'))
+                == sorted(old_metadata.data.get('labels'))
+            and new_metadata.data.get('policy_violation').get('name')
+                == old_metadata.data.get('policy_violation').get('name')
+        ):
+            # found the same license in existing entry, independent of the component-/
+            # resource-/package-version, so we must re-use its discovery date
+            return old_metadata.discovery_date
+
     elif new_metadata.type == odg.model.Datatype.DIKI_FINDING:
         if (
             new_metadata.data.get('provider_id') == old_metadata.data.get('provider_id')

odg/model.py

@@ -397,7 +397,7 @@ class BDBAMixin:
 
 
 @dataclasses.dataclass
-class PoliceViolationRef:
+class PolicyViolationRef:
     name: str
     id: str | None
     url: str | None
@@ -445,7 +445,7 @@ class IPFinding(Finding):
     package_name: str
     package_version: str | None
     license: License
-    policy_violation: PoliceViolationRef
+    policy_violation: PolicyViolationRef
     labels: list[str]
     host: str
 
@@ -482,6 +482,24 @@ def key(self) -> str:
         return _as_key(self.package_name, self.package_version, self.cve)
 
 
+@dataclasses.dataclass
+class RescoringIPFinding:
+    package_name: str
+    license: License
+    policy_violation: PolicyViolationRef
+    labels: list[str]
+
+    @property
+    def key(self) -> str:
+        labels_key = ','.join(sorted(self.labels))
+        return _as_key(
+            self.package_name,
+            self.license.name,
+            labels_key,
+            self.policy_violation.name
+        )
+
+
 @dataclasses.dataclass
 class RescoringVulnerabilityFinding:
     package_name: str
@@ -870,6 +888,7 @@ class CustomRescoring:
         | RescoringFalcoFinding
         | RescoringKyvernoFinding
         | RescoreGitHubSecretFinding
+        | RescoringIPFinding
     )
     referenced_type: str
     severity: str

rescore/artefacts.py

@@ -41,6 +41,15 @@ class LicenseFinding(odg.model.Finding):
     filesystem_paths: list[odg.model.FilesystemPath]
 
 
+@dataclasses.dataclass
+class IPFinding(odg.model.Finding):
+    package_name: str
+    package_versions: tuple[str, ...] # "..." for dacite.from_dict
+    license: odg.model.License
+    policy_violation: odg.model.PolicyViolationRef
+    labels: list[str]
+
+
 @dataclasses.dataclass
 class VulnerabilityFinding(odg.model.Finding):
     package_name: str
@@ -58,6 +67,7 @@ class RescoringProposal:
     finding: (
         LicenseFinding
         | VulnerabilityFinding
+        | IPFinding
         | odg.model.FindingModels
     )
     finding_type: odg.model.Datatype
@@ -312,6 +322,7 @@ async def _iter_rescoring_proposals(
         if finding_cfg.type in (
             odg.model.Datatype.VULNERABILITY_FINDING,
             odg.model.Datatype.LICENSE_FINDING,
+            odg.model.Datatype.IP_FINDING,
         ):
             artefact_metadata_with_same_ocm = tuple(
                 matching_am for matching_am in artefact_metadata
@@ -441,6 +452,51 @@ async def _iter_rescoring_proposals(
                         'sprint': sprint,
                     },
                 )
+            elif finding_cfg.type is odg.model.Datatype.IP_FINDING:
+                license = am.data.license
+
+                am_across_package_versions = tuple(
+                    matching_am for matching_am in artefact_metadata_with_same_ocm
+                    if (
+                        matching_am.data.license.name == license.name and
+                        matching_am.data.package_name == package_name and
+                        sorted(matching_am.data.labels) == sorted(am.data.labels) and
+                        matching_am.data.policy_violation.name == am.data.policy_violation.name
+                    )
+                )
+                seen_ids.update(
+                    tuple(
+                        local_am.id for local_am
+                        in am_across_package_versions
+                    )
+                )
+
+                package_versions, _ = _package_versions_and_filesystem_paths(
+                    artefact_metadata_across_package_version=am_across_package_versions,
+                    artefact_metadata=artefact_metadata,
+                    finding=am,
+                )
+
+                yield dacite.from_dict(
+                    data_class=RescoringProposal,
+                    data={
+                        'finding': {
+                            'package_name': package_name,
+                            'package_versions': package_versions,
+                            'severity': severity,
+                            'license': license,
+                            'policy_violation': am.data.policy_violation,
+                            'labels': am.data.labels,
+                        },
+                        'finding_type': finding_cfg.type,
+                        'severity': current_severity,
+                        'matching_rules': matching_rule_names,
+                        'applicable_rescorings': serialised_current_rescorings,
+                        'discovery_date': am.discovery_date.isoformat(),
+                        'due_date': due_date,
+                        'sprint': sprint,
+                    },
+                )
 
         else:
             yield dacite.from_dict(

rescore/utility.py

@@ -73,6 +73,14 @@ def _iter_rescorings_for_finding(
             ):
                 continue
 
+        elif finding.meta.type == odg.model.Datatype.IP_FINDING:
+            if (
+                rescoring.data.finding.license.name != finding.data.license.name
+                or rescoring.data.finding.package_name != finding.data.package_name
+                or rescoring.data.finding.policy_violation.name != finding.data.policy_violation.name
+                or sorted(rescoring.data.finding.labels) != sorted(finding.data.labels)
+            ):
+                continue
         else:
             if rescoring.data.finding.key != finding.data.key:
                 continue