threatbus/src/connector.py
def _map_to_threatbus(
        self, data: dict, opencti_action: str
    ) -> Union[Indicator, None]:
        """
        Inspects the given OpenCTI data point and either returns a valid STIX-2
        Indicator or None.
        @param data A dict object with OpenCTI SSE data
        @param opencti_action A string indicating what happened to this item
            (either `create`, `update` or `delete`)
        @return a STIX-2 Indicator or None
        """
        opencti_id: str = data.get("x_opencti_id", None)
        if not opencti_id:
            self.helper.log_error("Cannot process data without 'x_opencti_id' field")
            return
        indicator: dict = self.helper.api.indicator.read(id=opencti_id)
        if not indicator:
            # we are only interested in indicators at this time
            return
        # overwrite custom OpenCTI ID
        indicator["id"] = indicator.get("standard_id")
        if opencti_action == "update":
            indicator[
                ThreatBusSTIX2Constants.X_THREATBUS_UPDATE.value
            ] = Operation.EDIT.value
        if opencti_action == "delete":
            indicator[
                ThreatBusSTIX2Constants.X_THREATBUS_UPDATE.value
            ] = Operation.REMOVE.value
        return Indicator(**indicator, allow_custom=True)

        Inspects the given OpenCTI data point and either returns a valid STIX-2
        Indicator or None.
        @param data A dict object with OpenCTI SSE data
        @param opencti_action A string indicating what happened to this item
            (either `create`, `update` or `delete`)
        @return a STIX-2 Indicator or None
        
Inspects the given OpenCTI data point and either returns a valid STIX-2
Indicator or None.
Inspects the given OpenCTI data point and either returns a valid STIX - 2 Indicator or None .
def _map_to_threatbus(
        self, data: dict, opencti_action: str
    ) -> Union[Indicator, None]:
        opencti_id: str = data.get("x_opencti_id", None)
        if not opencti_id:
            self.helper.log_error("Cannot process data without 'x_opencti_id' field")
            return
        indicator: dict = self.helper.api.indicator.read(id=opencti_id)
        if not indicator:
            return
        indicator["id"] = indicator.get("standard_id")
        if opencti_action == "update":
            indicator[
                ThreatBusSTIX2Constants.X_THREATBUS_UPDATE.value
            ] = Operation.EDIT.value
        if opencti_action == "delete":
            indicator[
                ThreatBusSTIX2Constants.X_THREATBUS_UPDATE.value
            ] = Operation.REMOVE.value
        return Indicator(**indicator, allow_custom=True)
def _map_to_threatbus ( self , data : dict , opencti_action : str ) -> Union [ Indicator , None ] : opencti_id : str = data . get ( "x_opencti_id" , None ) if not opencti_id : self . helper . log_error ( "Cannot process data without 'x_opencti_id' field" ) return indicator : dict = self . helper . api . indicator . read ( id = opencti_id ) if not indicator : return indicator [ "id" ] = indicator . get ( "standard_id" ) if opencti_action == "update" : indicator [ ThreatBusSTIX2Constants . X_THREATBUS_UPDATE . value ] = Operation . EDIT . value if opencti_action == "delete" : indicator [ ThreatBusSTIX2Constants . X_THREATBUS_UPDATE . value ] = Operation . REMOVE . value return Indicator ( ** indicator , allow_custom = True )
Inspects the given OpenCTI data point and either returns a valid STIX-2
Indicator or None.
Inspects the given OpenCTI data point and either returns a valid STIX - 2 Indicator or None .
"""
        Inspects the given OpenCTI data point and either returns a valid STIX-2
        Indicator or None.
        @param data A dict object with OpenCTI SSE data
        @param opencti_action A string indicating what happened to this item
            (either `create`, `update` or `delete`)
        @return a STIX-2 Indicator or None
        """ # we are only interested in indicators at this time # overwrite custom OpenCTI ID
self
data
dict
opencti_action
str