(self)
Load data from files into data dict.
Load data from files into data dict.
Load data from files into data dict .
def load_data(self):
        """
        Load data from files into data dict.
        """
        if self.verbose:
            pbar = tqdm(total=len(self.types) * len(self.domains), desc="loading data", bar_format="{l_bar}{bar}| [{elapsed}<{remaining}, {rate_fmt}{postfix}]")
        for obj_type in self.types:
            for domain in self.domains:
                # handle data loaded from either a directory or the TAXII server
                def load_datastore(data_store):
                    raw_data = list(chain.from_iterable(
                        data_store.query(f) for f in attackTypeToStixFilter[obj_type]
                    ))
                    raw_data = self.deep_copy_stix(raw_data)
                    id_to_obj = {item['id']: item for item in raw_data}

                    return {
                        "id_to_obj": id_to_obj,
                        "keys": set(id_to_obj.keys()),
                        "data_store": data_store
                    }

                def parse_subtechniques(data_store, new=False):
                    # parse dataStore sub-technique-of relationships
                    if new: 
                        for technique in list(data_store.query(attackTypeToStixFilter["technique"])):
                            self.new_id_to_technique[technique["id"]] = technique
                        self.new_subtechnique_of_rels += list(data_store.query([
                            Filter("type", "=", "relationship"),
                            Filter("relationship_type", "=", "subtechnique-of")
                        ]))
                    else:
                        for technique in list(data_store.query(attackTypeToStixFilter["technique"])):
                            self.old_id_to_technique[technique["id"]] = technique
                        self.old_subtechnique_of_rels += list(data_store.query([
                            Filter("type", "=", "relationship"),
                            Filter("relationship_type", "=", "subtechnique-of")
                        ]))

                # load data from directory according to domain
                def load_dir(dir, new=False):
                    data_store = MemoryStore()
                    datafile = os.path.join(dir, domain + ".json")
                    data_store.load_from_file(datafile)
                    parse_subtechniques(data_store, new)
                    return load_datastore(data_store)

                # load data from TAXII server according to domain
                def load_taxii(new=False):
                    collection = Collection("https://cti-taxii.mitre.org/stix/collections/" + domainToTaxiiCollectionId[domain])
                    data_store = TAXIICollectionSource(collection)
                    parse_subtechniques(data_store, new)
                    return load_datastore(data_store)

                if self.use_taxii:
                    old = load_taxii(False)
                else:
                    old = load_dir(self.old, False)
                new = load_dir(self.new, True)

                intersection = old["keys"] & new["keys"]
                additions = new["keys"] - old["keys"]
                deletions = old["keys"] - new["keys"]

                # sets to store the ids of objects for each section
                changes = set()
                minor_changes = set()
                revocations = set()
                deprecations = set()
                unchanged = set()

                # find changes, revocations and deprecations
                for key in intersection:
                    if "revoked" in new["id_to_obj"][key] and new["id_to_obj"][key]["revoked"]:
                        if not "revoked" in old["id_to_obj"][key] or not old["id_to_obj"][key]["revoked"]: # if it was previously revoked, it's not a change
                            # store the revoking object
                            revoked_by_key = new["data_store"].query([
                                Filter('type', '=', 'relationship'),
                                Filter('relationship_type', '=', 'revoked-by'),
                                Filter('source_ref', '=', key)
                            ])
                            if (len(revoked_by_key) == 0): 
                                print("WARNING: revoked object", key, "has no revoked-by relationship")
                                continue
                            else: revoked_by_key = revoked_by_key[0]["target_ref"]

                            new["id_to_obj"][key]["revoked_by"] = new["id_to_obj"][revoked_by_key]

                            revocations.add(key)
                        # else it was already revoked, and not a change; do nothing with it
                    elif "x_mitre_deprecated" in new["id_to_obj"][key] and new["id_to_obj"][key]["x_mitre_deprecated"]:
                        if not "x_mitre_deprecated" in old["id_to_obj"][key]:   # if previously deprecated, not a change
                            deprecations.add(key)
                    else: # not revoked or deprecated
                        # try getting version numbers; should only lack version numbers if something has gone
                        # horribly wrong or a revoked object has slipped through
                        try:
                            old_version = float(old["id_to_obj"][key]["x_mitre_version"])
                        except: 
                            print("ERROR: cannot get old version for object: " + key)
                        try:
                            new_version = float(new["id_to_obj"][key]["x_mitre_version"])
                        except: 
                            print("ERROR: cannot get new version for object: " + key)

                        # check for changes
                        if new_version > old_version:
                            # an update has occurred to this object
                            changes.add(key)
                        else:
                            # check for minor change; modification date increased but not version
                            old_date = dateparser.parse(old["id_to_obj"][key]["modified"])
                            new_date = dateparser.parse(new["id_to_obj"][key]["modified"])
                            if new_date > old_date:
                                minor_changes.add(key)
                            else :
                                unchanged.add(key)
                
                # set data
                if obj_type not in self.data: self.data[obj_type] = {}
                self.data[obj_type][domain] = {
                    "additions":     [new["id_to_obj"][key] for key in additions],
                    "changes":       [new["id_to_obj"][key] for key in changes]
                }
                # only create minor_changes data if we want to display it later
                if self.minor_changes:
                    self.data[obj_type][domain]["minor_changes"] = [new["id_to_obj"][key] for key in minor_changes]
                
                # ditto for unchanged
                if self.unchanged:
                    self.data[obj_type][domain]["unchanged"] = [new["id_to_obj"][key] for key in unchanged]

                self.data[obj_type][domain]["revocations"] = [new["id_to_obj"][key] for key in revocations]
                self.data[obj_type][domain]["deprecations"] = [new["id_to_obj"][key] for key in deprecations]
                # only show deletions if objects were deleted
                if len(deletions) > 0:
                    self.data[obj_type][domain]["deletions"] = [old["id_to_obj"][key] for key in deletions]
                if self.verbose:
                    pbar.update(1)
        if self.verbose:
            pbar.close()
def load_data ( self ) : if self . verbose : pbar = tqdm ( total = len ( self . types ) * len ( self . domains ) , desc = "loading data" , bar_format = "{l_bar}{bar}| [{elapsed}<{remaining}, {rate_fmt}{postfix}]" ) for obj_type in self . types : for domain in self . domains : # handle data loaded from either a directory or the TAXII server def load_datastore ( data_store ) : raw_data = list ( chain . from_iterable ( data_store . query ( f ) for f in attackTypeToStixFilter [ obj_type ] ) ) raw_data = self . deep_copy_stix ( raw_data ) id_to_obj = { item [ 'id' ] : item for item in raw_data } return { "id_to_obj" : id_to_obj , "keys" : set ( id_to_obj . keys ( ) ) , "data_store" : data_store } def parse_subtechniques ( data_store , new = False ) : # parse dataStore sub-technique-of relationships if new : for technique in list ( data_store . query ( attackTypeToStixFilter [ "technique" ] ) ) : self . new_id_to_technique [ technique [ "id" ] ] = technique self . new_subtechnique_of_rels += list ( data_store . query ( [ Filter ( "type" , "=" , "relationship" ) , Filter ( "relationship_type" , "=" , "subtechnique-of" ) ] ) ) else : for technique in list ( data_store . query ( attackTypeToStixFilter [ "technique" ] ) ) : self . old_id_to_technique [ technique [ "id" ] ] = technique self . old_subtechnique_of_rels += list ( data_store . query ( [ Filter ( "type" , "=" , "relationship" ) , Filter ( "relationship_type" , "=" , "subtechnique-of" ) ] ) ) # load data from directory according to domain def load_dir ( dir , new = False ) : data_store = MemoryStore ( ) datafile = os . path . join ( dir , domain + ".json" ) data_store . load_from_file ( datafile ) parse_subtechniques ( data_store , new ) return load_datastore ( data_store ) # load data from TAXII server according to domain def load_taxii ( new = False ) : collection = Collection ( "https://cti-taxii.mitre.org/stix/collections/" + domainToTaxiiCollectionId [ domain ] ) data_store = TAXIICollectionSource ( collection ) parse_subtechniques ( data_store , new ) return load_datastore ( data_store ) if self . use_taxii : old = load_taxii ( False ) else : old = load_dir ( self . old , False ) new = load_dir ( self . new , True ) intersection = old [ "keys" ] & new [ "keys" ] additions = new [ "keys" ] - old [ "keys" ] deletions = old [ "keys" ] - new [ "keys" ] # sets to store the ids of objects for each section changes = set ( ) minor_changes = set ( ) revocations = set ( ) deprecations = set ( ) unchanged = set ( ) # find changes, revocations and deprecations for key in intersection : if "revoked" in new [ "id_to_obj" ] [ key ] and new [ "id_to_obj" ] [ key ] [ "revoked" ] : if not "revoked" in old [ "id_to_obj" ] [ key ] or not old [ "id_to_obj" ] [ key ] [ "revoked" ] : # if it was previously revoked, it's not a change # store the revoking object revoked_by_key = new [ "data_store" ] . query ( [ Filter ( 'type' , '=' , 'relationship' ) , Filter ( 'relationship_type' , '=' , 'revoked-by' ) , Filter ( 'source_ref' , '=' , key ) ] ) if ( len ( revoked_by_key ) == 0 ) : print ( "WARNING: revoked object" , key , "has no revoked-by relationship" ) continue else : revoked_by_key = revoked_by_key [ 0 ] [ "target_ref" ] new [ "id_to_obj" ] [ key ] [ "revoked_by" ] = new [ "id_to_obj" ] [ revoked_by_key ] revocations . add ( key ) # else it was already revoked, and not a change; do nothing with it elif "x_mitre_deprecated" in new [ "id_to_obj" ] [ key ] and new [ "id_to_obj" ] [ key ] [ "x_mitre_deprecated" ] : if not "x_mitre_deprecated" in old [ "id_to_obj" ] [ key ] : # if previously deprecated, not a change deprecations . add ( key ) else : # not revoked or deprecated # try getting version numbers; should only lack version numbers if something has gone # horribly wrong or a revoked object has slipped through try : old_version = float ( old [ "id_to_obj" ] [ key ] [ "x_mitre_version" ] ) except : print ( "ERROR: cannot get old version for object: " + key ) try : new_version = float ( new [ "id_to_obj" ] [ key ] [ "x_mitre_version" ] ) except : print ( "ERROR: cannot get new version for object: " + key ) # check for changes if new_version > old_version : # an update has occurred to this object changes . add ( key ) else : # check for minor change; modification date increased but not version old_date = dateparser . parse ( old [ "id_to_obj" ] [ key ] [ "modified" ] ) new_date = dateparser . parse ( new [ "id_to_obj" ] [ key ] [ "modified" ] ) if new_date > old_date : minor_changes . add ( key ) else : unchanged . add ( key ) # set data if obj_type not in self . data : self . data [ obj_type ] = { } self . data [ obj_type ] [ domain ] = { "additions" : [ new [ "id_to_obj" ] [ key ] for key in additions ] , "changes" : [ new [ "id_to_obj" ] [ key ] for key in changes ] } # only create minor_changes data if we want to display it later if self . minor_changes : self . data [ obj_type ] [ domain ] [ "minor_changes" ] = [ new [ "id_to_obj" ] [ key ] for key in minor_changes ] # ditto for unchanged if self . unchanged : self . data [ obj_type ] [ domain ] [ "unchanged" ] = [ new [ "id_to_obj" ] [ key ] for key in unchanged ] self . data [ obj_type ] [ domain ] [ "revocations" ] = [ new [ "id_to_obj" ] [ key ] for key in revocations ] self . data [ obj_type ] [ domain ] [ "deprecations" ] = [ new [ "id_to_obj" ] [ key ] for key in deprecations ] # only show deletions if objects were deleted if len ( deletions ) > 0 : self . data [ obj_type ] [ domain ] [ "deletions" ] = [ old [ "id_to_obj" ] [ key ] for key in deletions ] if self . verbose : pbar . update ( 1 ) if self . verbose : pbar . close ( )