(self, indicators, parent_description='')
Parse list of indicators.

        :param indicators: List of STIX indicators.
        :type indicators: List of STIX indicators.
        :param parent_description: The description of the parent indicator.
        :type parent_description: str.
Parse list of indicators.
Parse list of indicators .
def parse_indicators(self, indicators, parent_description=''):
        """
        Parse list of indicators.

        :param indicators: List of STIX indicators.
        :type indicators: List of STIX indicators.
        :param parent_description: The description of the parent indicator.
        :type parent_description: str.
        """

        for indicator in indicators: # for each STIX indicator
            if indicator.composite_indicator_expression: # parse indicator composition.
                # CRITs doesn't support complex boolean relationships like
                # ((A OR B) AND C). This code simply imports all indicators
                # and forms "Related_To" relationships between them

                # grab description while keeping any parent description
                p_description = ''
                if parent_description:
                    p_description = parent_description + '\n'
                if indicator.indicator_types:
                    for stix_ind_type in indicator.indicator_types:
                        p_description += ('STIX Indicator Type: ' +
                                          stix_ind_type.value + '\n')
                if indicator.description:
                    p_description += indicator.description.value

                cie = indicator.composite_indicator_expression
                self.parse_indicators(cie, p_description)
                rel_ids = []

                if self.preview: # no need to store relationship if just a preview
                    continue

                for com_ind in cie:
                    if com_ind.observables: # this contains an indicator
                        rel_ids.append(com_ind.id_)
                    else: # this contains another composition
                        if com_ind.id_ in self.idMap:
                            rel_ids.extend(self.idMap.pop(com_ind.id_))
                        else:
                            rel_ids.append(com_ind.id_)
                self.idMap.setdefault(indicator.id_, []).extend(rel_ids)

                # This is the top level, so form relationships
                if not isinstance(indicators, CompositeIndicatorExpression):
                    for iid in rel_ids:
                        for iid2 in rel_ids:
                            if iid != iid2:
                                self.relationships.append((iid,
                                                           RelationshipTypes.RELATED_TO,
                                                           iid2, "High"))
                continue

            # store relationships
            if not self.preview:
                for rel in getattr(indicator, 'related_indicators', ()) or ():
                    if rel.confidence:
                        conf = rel.confidence.value.value
                    else:
                        conf = 'Unknown'
                    self.relationships.append((indicator.id_,
                                               rel.relationship.value,
                                               rel.item.idref,
                                               conf))

            try: # create CRITs Indicator from observable
                description = [parent_description]
                # handled indicator-wrapped observable
                if getattr(indicator, 'title', ""):
                    if "Top-Level Object" in indicator.title:
                        self.parse_observables(indicator.observables,
                                               ind_id=indicator.id_)
                        continue
                    elif indicator.title:
                        title = indicator.title
                        description.append('STIX Indicator Title: %s' % title)
                if indicator.indicator_types and indicator.indicator_types[0]:
                    itype = indicator.indicator_types[0]
                    description.append('STIX Indicator Type: %s' % itype)
                if indicator.description:
                    desc = indicator.description
                    description.append('STIX Indicator Description: %s' % desc)
                description = '\n'.join(x for x in description if x)

                ci_vals = IndicatorCI.values()
                if (indicator.confidence
                    and indicator.confidence.value.value.lower() in ci_vals):
                    conf = indicator.confidence.value.value.lower()
                else:
                    conf = self.def_ci[0]
                if (indicator.likely_impact
                    and indicator.likely_impact.value.value.lower() in ci_vals):
                    impact = indicator.likely_impact.value.value.lower()
                else:
                    impact = self.def_ci[1]

                self.parse_observables(indicator.observables, description,
                                       True, indicator.id_, (conf, impact))

            except Exception as e:
                self.failed.append((e.message or str(e),
                                    "Indicator (%s)" % indicator.id_,
                                    indicator.id_)) # note for display in UI

            if indicator.indicated_ttps: # These aren't currently supported
                self.parse_ttps(indicator.indicated_ttps)

            if indicator.test_mechanisms:
                for tmech in indicator.test_mechanisms:
                    if isinstance(tmech, GenericTestMechanism):
                        kind = 'GENERIC'
                    elif isinstance (tmech, OpenIOCTestMechanism):
                        kind = 'OPEN IOC'
                    elif isinstance (tmech, SnortTestMechanism):
                        kind = 'SNORT'
                    elif isinstance (tmech, YaraTestMechanism):
                        kind = 'YARA'

                    ref = ', '.join(tmech.producer.references)

                    for rule in tmech.rules:
                        if not self.preview and user.has_access_to(SignatureACL.WRITE):
                            analyst = self.source_instance.analyst
                            res = handle_signature_file(str(rule),
                                                        self.source.name,
                                                        analyst,
                                                        title=tmech.id_,
                                                        data_type=kind.title(),
                                                        method='STIX Import',
                                                        reference=ref)
                            self.parsed.append(tmech.id_)
                            if res['success']:
                                oid = res['_id']
                                self.updates[oid] = res['object']
                                self.imported[tmech.id_] = (Signature._meta['crits_type'],
                                                            oid, "%s - %s" % (kind, rule))
                                self.relationships.append((tmech.id_,
                                                           RelationshipTypes.RELATED_TO,
                                                           indicator.id_, "High"))
                            elif 'Invalid data type' in res['message']:
                                msg = 'Add new Signature Type "%s" and try again'
                                self.failed.append((msg % kind.title(),
                                                    "Signature (%s)" % tmech.id_,
                                                    tmech.id_)) # note for display in UI
                            else:
                                self.failed.append((res['message'],
                                                    "Signature (%s)" % tmech.id_,
                                                    tmech.id_)) # note for display in UI
                        else: # preview only
                            stype = SignatureType.objects(name=kind.title()).first()
                            if not stype:
                                 msg = 'Add new Signature Type "%s" and try again'
                                 self.failed.append((msg % kind.title(),
                                                    "Signature (%s)" % tmech.id_,
                                                    tmech.id_)) # note for display in UI
                            else:
                                self.imported[tmech.id_] = (Signature._meta['crits_type'],
                                                            None, "%s - %s" % (kind, rule))
def parse_indicators ( self , indicators , parent_description = '' ) : for indicator in indicators : # for each STIX indicator if indicator . composite_indicator_expression : # parse indicator composition. # CRITs doesn't support complex boolean relationships like # ((A OR B) AND C). This code simply imports all indicators # and forms "Related_To" relationships between them # grab description while keeping any parent description p_description = '' if parent_description : p_description = parent_description + '\n' if indicator . indicator_types : for stix_ind_type in indicator . indicator_types : p_description += ( 'STIX Indicator Type: ' + stix_ind_type . value + '\n' ) if indicator . description : p_description += indicator . description . value cie = indicator . composite_indicator_expression self . parse_indicators ( cie , p_description ) rel_ids = [ ] if self . preview : # no need to store relationship if just a preview continue for com_ind in cie : if com_ind . observables : # this contains an indicator rel_ids . append ( com_ind . id_ ) else : # this contains another composition if com_ind . id_ in self . idMap : rel_ids . extend ( self . idMap . pop ( com_ind . id_ ) ) else : rel_ids . append ( com_ind . id_ ) self . idMap . setdefault ( indicator . id_ , [ ] ) . extend ( rel_ids ) # This is the top level, so form relationships if not isinstance ( indicators , CompositeIndicatorExpression ) : for iid in rel_ids : for iid2 in rel_ids : if iid != iid2 : self . relationships . append ( ( iid , RelationshipTypes . RELATED_TO , iid2 , "High" ) ) continue # store relationships if not self . preview : for rel in getattr ( indicator , 'related_indicators' , ( ) ) or ( ) : if rel . confidence : conf = rel . confidence . value . value else : conf = 'Unknown' self . relationships . append ( ( indicator . id_ , rel . relationship . value , rel . item . idref , conf ) ) try : # create CRITs Indicator from observable description = [ parent_description ] # handled indicator-wrapped observable if getattr ( indicator , 'title' , "" ) : if "Top-Level Object" in indicator . title : self . parse_observables ( indicator . observables , ind_id = indicator . id_ ) continue elif indicator . title : title = indicator . title description . append ( 'STIX Indicator Title: %s' % title ) if indicator . indicator_types and indicator . indicator_types [ 0 ] : itype = indicator . indicator_types [ 0 ] description . append ( 'STIX Indicator Type: %s' % itype ) if indicator . description : desc = indicator . description description . append ( 'STIX Indicator Description: %s' % desc ) description = '\n' . join ( x for x in description if x ) ci_vals = IndicatorCI . values ( ) if ( indicator . confidence and indicator . confidence . value . value . lower ( ) in ci_vals ) : conf = indicator . confidence . value . value . lower ( ) else : conf = self . def_ci [ 0 ] if ( indicator . likely_impact and indicator . likely_impact . value . value . lower ( ) in ci_vals ) : impact = indicator . likely_impact . value . value . lower ( ) else : impact = self . def_ci [ 1 ] self . parse_observables ( indicator . observables , description , True , indicator . id_ , ( conf , impact ) ) except Exception as e : self . failed . append ( ( e . message or str ( e ) , "Indicator (%s)" % indicator . id_ , indicator . id_ ) ) # note for display in UI if indicator . indicated_ttps : # These aren't currently supported self . parse_ttps ( indicator . indicated_ttps ) if indicator . test_mechanisms : for tmech in indicator . test_mechanisms : if isinstance ( tmech , GenericTestMechanism ) : kind = 'GENERIC' elif isinstance ( tmech , OpenIOCTestMechanism ) : kind = 'OPEN IOC' elif isinstance ( tmech , SnortTestMechanism ) : kind = 'SNORT' elif isinstance ( tmech , YaraTestMechanism ) : kind = 'YARA' ref = ', ' . join ( tmech . producer . references ) for rule in tmech . rules : if not self . preview and user . has_access_to ( SignatureACL . WRITE ) : analyst = self . source_instance . analyst res = handle_signature_file ( str ( rule ) , self . source . name , analyst , title = tmech . id_ , data_type = kind . title ( ) , method = 'STIX Import' , reference = ref ) self . parsed . append ( tmech . id_ ) if res [ 'success' ] : oid = res [ '_id' ] self . updates [ oid ] = res [ 'object' ] self . imported [ tmech . id_ ] = ( Signature . _meta [ 'crits_type' ] , oid , "%s - %s" % ( kind , rule ) ) self . relationships . append ( ( tmech . id_ , RelationshipTypes . RELATED_TO , indicator . id_ , "High" ) ) elif 'Invalid data type' in res [ 'message' ] : msg = 'Add new Signature Type "%s" and try again' self . failed . append ( ( msg % kind . title ( ) , "Signature (%s)" % tmech . id_ , tmech . id_ ) ) # note for display in UI else : self . failed . append ( ( res [ 'message' ] , "Signature (%s)" % tmech . id_ , tmech . id_ ) ) # note for display in UI else : # preview only stype = SignatureType . objects ( name = kind . title ( ) ) . first ( ) if not stype : msg = 'Add new Signature Type "%s" and try again' self . failed . append ( ( msg % kind . title ( ) , "Signature (%s)" % tmech . id_ , tmech . id_ ) ) # note for display in UI else : self . imported [ tmech . id_ ] = ( Signature . _meta [ 'crits_type' ] , None , "%s - %s" % ( kind , rule ) )