[client] Start renaming coverage (#opencti/11707)

Author: richard-julien

pycti/api/opencti_api_client.py

@@ -52,7 +52,7 @@
 from pycti.entities.opencti_opinion import Opinion
 from pycti.entities.opencti_report import Report
 from pycti.entities.opencti_role import Role
-from pycti.entities.opencti_security_assessment import SecurityAssessment
+from pycti.entities.opencti_security_coverage import SecurityCoverage
 from pycti.entities.opencti_settings import Settings
 from pycti.entities.opencti_stix import Stix
 from pycti.entities.opencti_stix_core_object import StixCoreObject
@@ -224,7 +224,7 @@ def __init__(
         self.narrative = Narrative(self)
         self.language = Language(self)
         self.vulnerability = Vulnerability(self)
-        self.security_assessment = SecurityAssessment(self)
+        self.security_coverage = SecurityCoverage(self)
         self.attack_pattern = AttackPattern(self)
         self.course_of_action = CourseOfAction(self)
         self.data_component = DataComponent(self)

…/entities/opencti_security_assessment.py …ti/entities/opencti_security_coverage.pypycti/entities/opencti_security_assessment.py renamed to pycti/entities/opencti_security_coverage.py pycti/entities/opencti_security_assessment.py renamed to pycti/entities/opencti_security_coverage.py

@@ -6,7 +6,7 @@
 from stix2.canonicalization.Canonicalize import canonicalize
 
 
-class SecurityAssessment:
+class SecurityCoverage:
     def __init__(self, opencti):
         self.opencti = opencti
         self.properties = """
@@ -17,8 +17,11 @@ def __init__(self, opencti):
             spec_version
             created_at
             updated_at
-            objectAssess {
-                id
+            objectCovered {
+                __typename
+                ... on StixCoreObject {
+                  id
+                }
             }
             objectMarking {
                 id
@@ -34,25 +37,24 @@ def __init__(self, opencti):
         """
 
     @staticmethod
-    def generate_id(name):
-        name = name.lower().strip()
-        data = {"name": name}
+    def generate_id(covered_ref):
+        data = {"covered_ref": covered_ref.lower().strip()}
         data = canonicalize(data, utf8=False)
         id = str(uuid.uuid5(uuid.UUID("00abedb4-aa42-466c-9c01-fed23315a9b7"), data))
-        return "securityAssessment--" + id
+        return "security-coverage--" + id
 
     @staticmethod
     def generate_id_from_data(data):
-        return SecurityAssessment.generate_id(data["name"])
+        return SecurityCoverage.generate_id(data["covered_ref"])
 
     """
-        List SecurityAssessment objects
+        List securityCoverage objects
 
         :param filters: the filters to apply
         :param search: the search keyword
         :param first: return the first n rows from the after ID (or the beginning if not set)
         :param after: ID of the first row for pagination
-        :return List of SecurityAssessment objects
+        :return List of SecurityCoverage objects
     """
 
     def list(self, **kwargs):
@@ -67,12 +69,12 @@ def list(self, **kwargs):
         with_pagination = kwargs.get("withPagination", False)
 
         self.opencti.app_logger.info(
-            "Listing SecurityAssessment with filters", {"filters": json.dumps(filters)}
+            "Listing SecurityCoverage with filters", {"filters": json.dumps(filters)}
         )
         query = (
             """
-                query SecurityAssessment($filters: FilterGroup, $search: String, $first: Int, $after: ID, $orderBy: SecurityAssessmentOrdering, $orderMode: OrderingMode) {
-                    securityAssessments(filters: $filters, search: $search, first: $first, after: $after, orderBy: $orderBy, orderMode: $orderMode) {
+                query SecurityCoverage($filters: FilterGroup, $search: String, $first: Int, $after: ID, $orderBy: SecurityCoverageOrdering, $orderMode: OrderingMode) {
+                    securityCoverages(filters: $filters, search: $search, first: $first, after: $after, orderBy: $orderBy, orderMode: $orderMode) {
                         edges {
                             node {
                                 """
@@ -105,12 +107,12 @@ def list(self, **kwargs):
 
         if get_all:
             final_data = []
-            data = self.opencti.process_multiple(result["data"]["securityAssessments"])
+            data = self.opencti.process_multiple(result["data"]["securityCoverages"])
             final_data = final_data + data
-            while result["data"]["securityAssessments"]["pageInfo"]["hasNextPage"]:
-                after = result["data"]["securityAssessments"]["pageInfo"]["endCursor"]
+            while result["data"]["securityCoverages"]["pageInfo"]["hasNextPage"]:
+                after = result["data"]["securityCoverages"]["pageInfo"]["endCursor"]
                 self.opencti.app_logger.info(
-                    "Listing SecurityAssessment", {"after": after}
+                    "Listing SecurityCoverage", {"after": after}
                 )
                 result = self.opencti.query(
                     query,
@@ -124,33 +126,33 @@ def list(self, **kwargs):
                     },
                 )
                 data = self.opencti.process_multiple(
-                    result["data"]["securityAssessments"]
+                    result["data"]["securityCoverages"]
                 )
                 final_data = final_data + data
             return final_data
         else:
             return self.opencti.process_multiple(
-                result["data"]["securityAssessments"], with_pagination
+                result["data"]["securityCoverages"], with_pagination
             )
 
     """
-        Read a SecurityAssessment object
+        Read a SecurityCoverage object
 
-        :param id: the id of the SecurityAssessment
+        :param id: the id of the SecurityCoverage
         :param filters: the filters to apply if no id provided
-        :return SecurityAssessment object
+        :return SecurityCoverage object
     """
 
     def read(self, **kwargs):
         id = kwargs.get("id", None)
         filters = kwargs.get("filters", None)
         custom_attributes = kwargs.get("customAttributes", None)
         if id is not None:
-            self.opencti.app_logger.info("Reading SecurityAssessment", {"id": id})
+            self.opencti.app_logger.info("Reading SecurityCoverage", {"id": id})
             query = (
                 """
-                    query SecurityAssessment($id: String!) {
-                        securityAssessment(id: $id) {
+                    query SecurityCoverage($id: String!) {
+                        securityCoverage(id: $id) {
                             """
                 + (
                     custom_attributes
@@ -164,7 +166,7 @@ def read(self, **kwargs):
             )
             result = self.opencti.query(query, {"id": id})
             return self.opencti.process_multiple_fields(
-                result["data"]["securityAssessment"]
+                result["data"]["securityCoverage"]
             )
         elif filters is not None:
             result = self.list(filters=filters)
@@ -174,6 +176,6 @@ def read(self, **kwargs):
                 return None
         else:
             self.opencti.app_logger.error(
-                "[opencti_tool] Missing parameters: id or filters"
+                "[opencti_security_coverage] Missing parameters: id or filters"
             )
             return None

pycti/utils/opencti_stix2.py

@@ -885,7 +885,7 @@ def get_readers(self):
             "Tool": self.opencti.tool.read,
             "Vocabulary": self.opencti.vocabulary.read,
             "Vulnerability": self.opencti.vulnerability.read,
-            "SecurityAssessment": self.opencti.security_assessment.read,
+            "Security-Coverage": self.opencti.security_coverage.read,
         }
 
     def get_reader(self, entity_type: str):
@@ -962,6 +962,7 @@ def get_stix_helper(self):
             "narrative": self.opencti.narrative,
             "task": self.opencti.task,
             "x-opencti-task": self.opencti.task,
+            "Security-Coverage": self.opencti.security_coverage,
             "vocabulary": self.opencti.vocabulary,
             # relationships
             "relationship": self.opencti.stix_core_relationship,