[client] Introduce security assessment (opencti/#11707)

Author: richard-julien

pycti/api/opencti_api_client.py

@@ -52,6 +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_settings import Settings
 from pycti.entities.opencti_stix import Stix
 from pycti.entities.opencti_stix_core_object import StixCoreObject
@@ -223,6 +224,7 @@ def __init__(
         self.narrative = Narrative(self)
         self.language = Language(self)
         self.vulnerability = Vulnerability(self)
+        self.security_assessment = SecurityAssessment(self)
         self.attack_pattern = AttackPattern(self)
         self.course_of_action = CourseOfAction(self)
         self.data_component = DataComponent(self)

pycti/connector/opencti_connector.py

@@ -43,6 +43,8 @@ def __init__(
         auto: bool,
         only_contextual: bool,
         playbook_compatible: bool,
+        auto_update: bool,
+        enrichment_resolution: str,
         listen_callback_uri=None,
     ):
         self.id = connector_id
@@ -55,6 +57,8 @@ def __init__(
         else:
             self.scope = []
         self.auto = auto
+        self.auto_update = auto_update
+        self.enrichment_resolution = enrichment_resolution
         self.only_contextual = only_contextual
         self.playbook_compatible = playbook_compatible
         self.listen_callback_uri = listen_callback_uri
@@ -72,6 +76,8 @@ def to_input(self) -> dict:
                 "type": self.type.name,
                 "scope": self.scope,
                 "auto": self.auto,
+                "auto_update": self.auto_update,
+                "enrichment_resolution": self.enrichment_resolution,
                 "only_contextual": self.only_contextual,
                 "playbook_compatible": self.playbook_compatible,
                 "listen_callback_uri": self.listen_callback_uri,

pycti/connector/opencti_connector_helper.py

@@ -365,6 +365,16 @@ def _data_handler(self, json_data) -> None:
             event_data = json_data["event"]
             entity_id = event_data.get("entity_id")
             entity_type = event_data.get("entity_type")
+            stix_entity = (
+                json.loads(event_data.get("stix_entity"))
+                if event_data.get("stix_entity")
+                else None
+            )
+            stix_objects = (
+                json.loads(event_data.get("stix_objects"))
+                if event_data.get("stix_objects")
+                else None
+            )
             validation_mode = event_data.get("validation_mode", "workbench")
             force_validation = event_data.get("force_validation", False)
             # Set the API headers
@@ -430,15 +440,16 @@ def _data_handler(self, json_data) -> None:
                 else:
                     # If not playbook but enrichment, compute object on enrichment_entity
                     opencti_entity = event_data["enrichment_entity"]
-                    stix_objects = self.helper.api.stix2.prepare_export(
-                        entity=self.helper.api.stix2.generate_export(
-                            copy.copy(opencti_entity)
+                    if stix_objects is None:
+                        stix_objects = self.helper.api.stix2.prepare_export(
+                            entity=self.helper.api.stix2.generate_export(
+                                copy.copy(opencti_entity)
+                            )
                         )
-                    )
-                    stix_entity = [
-                        e
-                        for e in stix_objects
-                        if e["id"] == opencti_entity["standard_id"]
+                        stix_entity = [
+                            e
+                            for e in stix_objects
+                            if e["id"] == opencti_entity["standard_id"]
                         or e["id"] == "x-opencti-" + opencti_entity["standard_id"]
                     ][0]
                     event_data["stix_objects"] = stix_objects
@@ -1116,6 +1127,15 @@ def __init__(self, config: Dict, playbook_compatible=False) -> None:
         self.connect_auto = get_config_variable(
             "CONNECTOR_AUTO", ["connector", "auto"], config, default=False
         )
+        self.connect_auto_update = get_config_variable(
+            "CONNECTOR_AUTO_UPDATE", ["connector", "auto_update"], config, default=False
+        )
+        self.connect_enrichment_resolution = get_config_variable(
+            "CONNECTOR_ENRICHMENT_RESOLUTION",
+            ["connector", "enrichment_resolution"],
+            config,
+            default="none",
+        )
         self.bundle_send_to_queue = get_config_variable(
             "CONNECTOR_SEND_TO_QUEUE",
             ["connector", "send_to_queue"],
@@ -1231,14 +1251,16 @@ def __init__(self, config: Dict, playbook_compatible=False) -> None:
         )
         # Register the connector in OpenCTI
         self.connector = OpenCTIConnector(
-            self.connect_id,
-            self.connect_name,
-            self.connect_type,
-            self.connect_scope,
-            self.connect_auto,
-            self.connect_only_contextual,
-            playbook_compatible,
-            (
+            connector_id=self.connect_id,
+            connector_name=self.connect_name,
+            connector_type=self.connect_type,
+            scope=self.connect_scope,
+            auto=self.connect_auto,
+            only_contextual=self.connect_only_contextual,
+            playbook_compatible=playbook_compatible,
+            auto_update=self.connect_auto_update,
+            enrichment_resolution=self.connect_enrichment_resolution,
+            listen_callback_uri=(
                 self.listen_protocol_api_uri + self.listen_protocol_api_path
                 if self.listen_protocol == "API"
                 else None

pycti/entities/opencti_security_assessment.py

@@ -0,0 +1,179 @@
+# coding: utf-8
+
+import json
+import uuid
+
+from stix2.canonicalization.Canonicalize import canonicalize
+
+
+class SecurityAssessment:
+    def __init__(self, opencti):
+        self.opencti = opencti
+        self.properties = """
+            id
+            standard_id
+            entity_type
+            parent_types
+            spec_version
+            created_at
+            updated_at
+            objectAssess {
+                id
+            }
+            objectMarking {
+                id
+                standard_id
+                entity_type
+                definition_type
+                definition
+                created
+                modified
+                x_opencti_order
+                x_opencti_color
+            }
+        """
+
+    @staticmethod
+    def generate_id(name):
+        name = name.lower().strip()
+        data = {"name": name}
+        data = canonicalize(data, utf8=False)
+        id = str(uuid.uuid5(uuid.UUID("00abedb4-aa42-466c-9c01-fed23315a9b7"), data))
+        return "securityAssessment--" + id
+
+    @staticmethod
+    def generate_id_from_data(data):
+        return SecurityAssessment.generate_id(data["name"])
+
+    """
+        List SecurityAssessment 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
+    """
+
+    def list(self, **kwargs):
+        filters = kwargs.get("filters", None)
+        search = kwargs.get("search", None)
+        first = kwargs.get("first", 100)
+        after = kwargs.get("after", None)
+        order_by = kwargs.get("orderBy", None)
+        order_mode = kwargs.get("orderMode", None)
+        custom_attributes = kwargs.get("customAttributes", None)
+        get_all = kwargs.get("getAll", False)
+        with_pagination = kwargs.get("withPagination", False)
+
+        self.opencti.app_logger.info(
+            "Listing SecurityAssessment 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) {
+                        edges {
+                            node {
+                                """
+            + (custom_attributes if custom_attributes is not None else self.properties)
+            + """
+                        }
+                    }
+                    pageInfo {
+                        startCursor
+                        endCursor
+                        hasNextPage
+                        hasPreviousPage
+                        globalCount
+                    }
+                }
+            }
+        """
+        )
+        result = self.opencti.query(
+            query,
+            {
+                "filters": filters,
+                "search": search,
+                "first": first,
+                "after": after,
+                "orderBy": order_by,
+                "orderMode": order_mode,
+            },
+        )
+
+        if get_all:
+            final_data = []
+            data = self.opencti.process_multiple(result["data"]["securityAssessments"])
+            final_data = final_data + data
+            while result["data"]["securityAssessments"]["pageInfo"]["hasNextPage"]:
+                after = result["data"]["securityAssessments"]["pageInfo"]["endCursor"]
+                self.opencti.app_logger.info(
+                    "Listing SecurityAssessment", {"after": after}
+                )
+                result = self.opencti.query(
+                    query,
+                    {
+                        "filters": filters,
+                        "search": search,
+                        "first": first,
+                        "after": after,
+                        "orderBy": order_by,
+                        "orderMode": order_mode,
+                    },
+                )
+                data = self.opencti.process_multiple(
+                    result["data"]["securityAssessments"]
+                )
+                final_data = final_data + data
+            return final_data
+        else:
+            return self.opencti.process_multiple(
+                result["data"]["securityAssessments"], with_pagination
+            )
+
+    """
+        Read a SecurityAssessment object
+
+        :param id: the id of the SecurityAssessment
+        :param filters: the filters to apply if no id provided
+        :return SecurityAssessment 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})
+            query = (
+                """
+                    query SecurityAssessment($id: String!) {
+                        securityAssessment(id: $id) {
+                            """
+                + (
+                    custom_attributes
+                    if custom_attributes is not None
+                    else self.properties
+                )
+                + """
+                    }
+                }
+             """
+            )
+            result = self.opencti.query(query, {"id": id})
+            return self.opencti.process_multiple_fields(
+                result["data"]["securityAssessment"]
+            )
+        elif filters is not None:
+            result = self.list(filters=filters)
+            if len(result) > 0:
+                return result[0]
+            else:
+                return None
+        else:
+            self.opencti.app_logger.error(
+                "[opencti_tool] Missing parameters: id or filters"
+            )
+            return None

pycti/utils/opencti_stix2.py

@@ -885,6 +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,
         }
 
     def get_reader(self, entity_type: str):