import json import os import pytest import stix2 import stix2.environment import stix2.equivalence.graph import stix2.equivalence.object import stix2.exceptions from .constants import ( ATTACK_PATTERN_ID, ATTACK_PATTERN_KWARGS, CAMPAIGN_ID, CAMPAIGN_KWARGS, FAKE_TIME, IDENTITY_ID, IDENTITY_KWARGS, INDICATOR_ID, INDICATOR_KWARGS, LOCATION_ID, LOCATION_KWARGS, MALWARE_ID, MALWARE_KWARGS, RELATIONSHIP_IDS, REPORT_ID, REPORT_KWARGS, THREAT_ACTOR_ID, THREAT_ACTOR_KWARGS, TOOL_ID, TOOL_KWARGS, VULNERABILITY_ID, VULNERABILITY_KWARGS, ) FS_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stix2_data") @pytest.fixture def ds(): cam = stix2.v21.Campaign(id=CAMPAIGN_ID, **CAMPAIGN_KWARGS) idy = stix2.v21.Identity(id=IDENTITY_ID, **IDENTITY_KWARGS) ind = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) mal = stix2.v21.Malware(id=MALWARE_ID, **MALWARE_KWARGS) rel1 = stix2.v21.Relationship(ind, 'indicates', mal, id=RELATIONSHIP_IDS[0]) rel2 = stix2.v21.Relationship(mal, 'targets', idy, id=RELATIONSHIP_IDS[1]) rel3 = stix2.v21.Relationship(cam, 'uses', mal, id=RELATIONSHIP_IDS[2]) reprt = stix2.v21.Report( name="Malware Report", published="2021-05-09T08:22:22Z", object_refs=[mal.id, rel1.id, ind.id], ) stix_objs = [cam, idy, ind, mal, rel1, rel2, rel3, reprt] yield stix2.MemoryStore(stix_objs) @pytest.fixture def ds2_objects(): cam = stix2.v21.Campaign(id=CAMPAIGN_ID, **CAMPAIGN_KWARGS) idy = stix2.v21.Identity(id=IDENTITY_ID, **IDENTITY_KWARGS) ind = stix2.v21.Indicator(id=INDICATOR_ID, created_by_ref=idy.id, **INDICATOR_KWARGS) indv2 = ind.new_version( external_references=[ { "source_name": "unknown", "url": "https://examplewebsite.com/", }, ], object_marking_refs=[stix2.v21.TLP_WHITE], ) mal = stix2.v21.Malware(id=MALWARE_ID, created_by_ref=idy.id, **MALWARE_KWARGS) malv2 = mal.new_version( external_references=[ { "source_name": "unknown", "url": "https://examplewebsite2.com/", }, ], ) rel1 = stix2.v21.Relationship(ind, 'indicates', mal, id=RELATIONSHIP_IDS[0]) rel2 = stix2.v21.Relationship(mal, 'targets', idy, id=RELATIONSHIP_IDS[1]) rel3 = stix2.v21.Relationship(cam, 'uses', mal, id=RELATIONSHIP_IDS[2]) stix_objs = [cam, idy, ind, indv2, mal, malv2, rel1, rel2, rel3] reprt = stix2.v21.Report( created_by_ref=idy.id, name="example", published="2021-04-09T08:22:22Z", object_refs=stix_objs, ) stix_objs.append(reprt) yield stix_objs @pytest.fixture def ds2(ds2_objects): yield stix2.MemoryStore(ds2_objects) @pytest.fixture def fs(): yield stix2.FileSystemSource(FS_PATH) def test_object_factory_created_by_ref_str(): factory = stix2.ObjectFactory(created_by_ref=IDENTITY_ID) ind = factory.create(stix2.v21.Indicator, **INDICATOR_KWARGS) assert ind.created_by_ref == IDENTITY_ID def test_object_factory_created_by_ref_obj(): id_obj = stix2.v21.Identity(id=IDENTITY_ID, **IDENTITY_KWARGS) factory = stix2.ObjectFactory(created_by_ref=id_obj) ind = factory.create(stix2.v21.Indicator, **INDICATOR_KWARGS) assert ind.created_by_ref == IDENTITY_ID def test_object_factory_override_default(): factory = stix2.ObjectFactory(created_by_ref=IDENTITY_ID) new_id = "identity--983b3172-44fe-4a80-8091-eb8098841fe8" ind = factory.create(stix2.v21.Indicator, created_by_ref=new_id, **INDICATOR_KWARGS) assert ind.created_by_ref == new_id def test_object_factory_created(): factory = stix2.ObjectFactory(created=FAKE_TIME) ind = factory.create(stix2.v21.Indicator, **INDICATOR_KWARGS) assert ind.created == FAKE_TIME assert ind.modified == FAKE_TIME def test_object_factory_external_reference(): ext_ref = stix2.v21.ExternalReference( source_name="ACME Threat Intel", description="Threat report", ) factory = stix2.ObjectFactory(external_references=ext_ref) ind = factory.create(stix2.v21.Indicator, **INDICATOR_KWARGS) assert ind.external_references[0].source_name == "ACME Threat Intel" assert ind.external_references[0].description == "Threat report" ind2 = factory.create(stix2.v21.Indicator, external_references=None, **INDICATOR_KWARGS) assert 'external_references' not in ind2 def test_object_factory_obj_markings(): stmt_marking = stix2.v21.StatementMarking("Copyright 2016, Example Corp") mark_def = stix2.v21.MarkingDefinition( definition_type="statement", definition=stmt_marking, ) factory = stix2.ObjectFactory(object_marking_refs=[mark_def, stix2.v21.TLP_AMBER]) ind = factory.create(stix2.v21.Indicator, **INDICATOR_KWARGS) assert mark_def.id in ind.object_marking_refs assert stix2.v21.TLP_AMBER.id in ind.object_marking_refs factory = stix2.ObjectFactory(object_marking_refs=stix2.v21.TLP_RED) ind = factory.create(stix2.v21.Indicator, **INDICATOR_KWARGS) assert stix2.v21.TLP_RED.id in ind.object_marking_refs def test_object_factory_list_append(): ext_ref = stix2.v21.ExternalReference( source_name="ACME Threat Intel", description="Threat report from ACME", ) ext_ref2 = stix2.v21.ExternalReference( source_name="Yet Another Threat Report", description="Threat report from YATR", ) ext_ref3 = stix2.v21.ExternalReference( source_name="Threat Report #3", description="One more threat report", ) factory = stix2.ObjectFactory(external_references=ext_ref) ind = factory.create(stix2.v21.Indicator, external_references=ext_ref2, **INDICATOR_KWARGS) assert ind.external_references[1].source_name == "Yet Another Threat Report" ind = factory.create(stix2.v21.Indicator, external_references=[ext_ref2, ext_ref3], **INDICATOR_KWARGS) assert ind.external_references[2].source_name == "Threat Report #3" def test_object_factory_list_replace(): ext_ref = stix2.v21.ExternalReference( source_name="ACME Threat Intel", description="Threat report from ACME", ) ext_ref2 = stix2.v21.ExternalReference( source_name="Yet Another Threat Report", description="Threat report from YATR", ) factory = stix2.ObjectFactory(external_references=ext_ref, list_append=False) ind = factory.create(stix2.v21.Indicator, external_references=ext_ref2, **INDICATOR_KWARGS) assert len(ind.external_references) == 1 assert ind.external_references[0].source_name == "Yet Another Threat Report" def test_environment_functions(): env = stix2.Environment( stix2.ObjectFactory(created_by_ref=IDENTITY_ID), stix2.MemoryStore(), ) # Create a STIX object ind = env.create(stix2.v21.Indicator, id=INDICATOR_ID, **INDICATOR_KWARGS) assert ind.created_by_ref == IDENTITY_ID # Add objects to datastore ind2 = ind.new_version(labels=['benign']) env.add([ind, ind2]) # Get both versions of the object resp = env.all_versions(INDICATOR_ID) assert len(resp) == 2 # Get just the most recent version of the object resp = env.get(INDICATOR_ID) assert resp['labels'][0] == 'benign' # Search on something other than id query = [stix2.Filter('type', '=', 'vulnerability')] resp = env.query(query) assert len(resp) == 0 # See different results after adding filters to the environment env.add_filters([ stix2.Filter('type', '=', 'indicator'), stix2.Filter('created_by_ref', '=', IDENTITY_ID), ]) env.add_filter(stix2.Filter('labels', '=', 'benign')) # should be 'malicious-activity' resp = env.get(INDICATOR_ID) assert resp['labels'][0] == 'benign' # should be 'malicious-activity' def test_environment_source_and_sink(): ind = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) env = stix2.Environment(source=stix2.MemorySource([ind]), sink=stix2.MemorySink([ind])) assert env.get(INDICATOR_ID).indicator_types[0] == 'malicious-activity' def test_environment_datastore_and_sink(): with pytest.raises(ValueError) as excinfo: stix2.Environment( factory=stix2.ObjectFactory(), store=stix2.MemoryStore(), sink=stix2.MemorySink, ) assert 'Data store already provided' in str(excinfo.value) def test_environment_no_datastore(): env = stix2.Environment(factory=stix2.ObjectFactory()) with pytest.raises(AttributeError) as excinfo: env.add(stix2.v21.Indicator(**INDICATOR_KWARGS)) assert 'Environment has no data sink to put objects in' in str(excinfo.value) with pytest.raises(AttributeError) as excinfo: env.get(INDICATOR_ID) assert 'Environment has no data source' in str(excinfo.value) with pytest.raises(AttributeError) as excinfo: env.all_versions(INDICATOR_ID) assert 'Environment has no data source' in str(excinfo.value) with pytest.raises(AttributeError) as excinfo: env.query(INDICATOR_ID) assert 'Environment has no data source' in str(excinfo.value) with pytest.raises(AttributeError) as excinfo: env.relationships(INDICATOR_ID) assert 'Environment has no data source' in str(excinfo.value) with pytest.raises(AttributeError) as excinfo: env.related_to(INDICATOR_ID) assert 'Environment has no data source' in str(excinfo.value) def test_environment_add_filters(): env = stix2.Environment(factory=stix2.ObjectFactory()) env.add_filters([INDICATOR_ID]) env.add_filter(INDICATOR_ID) def test_environment_datastore_and_no_object_factory(): # Uses a default object factory env = stix2.Environment(store=stix2.MemoryStore()) ind = env.create(stix2.v21.Indicator, id=INDICATOR_ID, **INDICATOR_KWARGS) assert ind.id == INDICATOR_ID def test_parse_malware(): env = stix2.Environment() data = """{ "type": "malware", "spec_version": "2.1", "id": "malware--9c4638ec-f1de-4ddb-abf4-1b760417654e", "created": "2017-01-01T12:34:56.000Z", "modified": "2017-01-01T12:34:56.000Z", "name": "Cryptolocker", "malware_types": [ "ransomware" ], "is_family": false }""" mal = env.parse(data, version="2.1") assert mal.type == 'malware' assert mal.spec_version == '2.1' assert mal.id == MALWARE_ID assert mal.created == FAKE_TIME assert mal.modified == FAKE_TIME assert mal.malware_types == ['ransomware'] assert mal.name == "Cryptolocker" assert not mal.is_family def test_creator_of(): identity = stix2.v21.Identity(**IDENTITY_KWARGS) factory = stix2.ObjectFactory(created_by_ref=identity.id) env = stix2.Environment(store=stix2.MemoryStore(), factory=factory) env.add(identity) ind = env.create(stix2.v21.Indicator, **INDICATOR_KWARGS) creator = env.creator_of(ind) assert creator is identity def test_creator_of_no_datasource(): identity = stix2.v21.Identity(**IDENTITY_KWARGS) factory = stix2.ObjectFactory(created_by_ref=identity.id) env = stix2.Environment(factory=factory) ind = env.create(stix2.v21.Indicator, **INDICATOR_KWARGS) with pytest.raises(AttributeError) as excinfo: env.creator_of(ind) assert 'Environment has no data source' in str(excinfo.value) def test_creator_of_not_found(): identity = stix2.v21.Identity(**IDENTITY_KWARGS) factory = stix2.ObjectFactory(created_by_ref=identity.id) env = stix2.Environment(store=stix2.MemoryStore(), factory=factory) ind = env.create(stix2.v21.Indicator, **INDICATOR_KWARGS) creator = env.creator_of(ind) assert creator is None def test_creator_of_no_created_by_ref(): env = stix2.Environment(store=stix2.MemoryStore()) ind = env.create(stix2.v21.Indicator, **INDICATOR_KWARGS) creator = env.creator_of(ind) assert creator is None def test_relationships(ds): env = stix2.Environment(store=ds) mal = env.get(MALWARE_ID) resp = env.relationships(mal) assert len(resp) == 3 assert any(x['id'] == RELATIONSHIP_IDS[0] for x in resp) assert any(x['id'] == RELATIONSHIP_IDS[1] for x in resp) assert any(x['id'] == RELATIONSHIP_IDS[2] for x in resp) def test_relationships_no_id(ds): env = stix2.Environment(store=ds) mal = { "type": "malware", "name": "some variant", } with pytest.raises(ValueError) as excinfo: env.relationships(mal) assert "object has no 'id' property" in str(excinfo.value) def test_relationships_by_type(ds): env = stix2.Environment(store=ds) mal = env.get(MALWARE_ID) resp = env.relationships(mal, relationship_type='indicates') assert len(resp) == 1 assert resp[0]['id'] == RELATIONSHIP_IDS[0] def test_relationships_by_source(ds): env = stix2.Environment(store=ds) resp = env.relationships(MALWARE_ID, source_only=True) assert len(resp) == 1 assert resp[0]['id'] == RELATIONSHIP_IDS[1] def test_relationships_by_target(ds): env = stix2.Environment(store=ds) resp = env.relationships(MALWARE_ID, target_only=True) assert len(resp) == 2 assert any(x['id'] == RELATIONSHIP_IDS[0] for x in resp) assert any(x['id'] == RELATIONSHIP_IDS[2] for x in resp) def test_relationships_by_target_and_type(ds): env = stix2.Environment(store=ds) resp = env.relationships(MALWARE_ID, relationship_type='uses', target_only=True) assert len(resp) == 1 assert any(x['id'] == RELATIONSHIP_IDS[2] for x in resp) def test_relationships_by_target_and_source(ds): env = stix2.Environment(store=ds) with pytest.raises(ValueError) as excinfo: env.relationships(MALWARE_ID, target_only=True, source_only=True) assert 'not both' in str(excinfo.value) def test_related_to(ds): env = stix2.Environment(store=ds) mal = env.get(MALWARE_ID) resp = env.related_to(mal) assert len(resp) == 3 assert any(x['id'] == CAMPAIGN_ID for x in resp) assert any(x['id'] == INDICATOR_ID for x in resp) assert any(x['id'] == IDENTITY_ID for x in resp) def test_related_to_no_id(ds): env = stix2.Environment(store=ds) mal = { "type": "malware", "name": "some variant", "is_family": False, } with pytest.raises(ValueError) as excinfo: env.related_to(mal) assert "object has no 'id' property" in str(excinfo.value) def test_related_to_by_source(ds): env = stix2.Environment(store=ds) resp = env.related_to(MALWARE_ID, source_only=True) assert len(resp) == 1 assert resp[0]['id'] == IDENTITY_ID def test_related_to_by_target(ds): env = stix2.Environment(store=ds) resp = env.related_to(MALWARE_ID, target_only=True) assert len(resp) == 2 assert any(x['id'] == CAMPAIGN_ID for x in resp) assert any(x['id'] == INDICATOR_ID for x in resp) def test_object_similarity_on_same_attack_pattern1(): ap1 = stix2.v21.AttackPattern(id=ATTACK_PATTERN_ID, **ATTACK_PATTERN_KWARGS) ap2 = stix2.v21.AttackPattern(id=ATTACK_PATTERN_ID, **ATTACK_PATTERN_KWARGS) env = stix2.Environment().object_similarity(ap1, ap2) assert round(env) == 100 def test_object_similarity_on_same_attack_pattern2(): ATTACK_KWARGS = dict( name="Phishing", external_references=[ { "url": "https://example2", "source_name": "some-source2", }, ], ) ap1 = stix2.v21.AttackPattern(id=ATTACK_PATTERN_ID, **ATTACK_KWARGS) ap2 = stix2.v21.AttackPattern(id=ATTACK_PATTERN_ID, **ATTACK_KWARGS) env = stix2.Environment().object_similarity(ap1, ap2) assert round(env) == 100 def test_object_similarity_on_same_campaign1(): camp1 = stix2.v21.Campaign(id=CAMPAIGN_ID, **CAMPAIGN_KWARGS) camp2 = stix2.v21.Campaign(id=CAMPAIGN_ID, **CAMPAIGN_KWARGS) env = stix2.Environment().object_similarity(camp1, camp2) assert round(env) == 100 def test_object_similarity_on_same_campaign2(): CAMP_KWARGS = dict( name="Green Group Attacks Against Finance", description="Campaign by Green Group against a series of targets in the financial services sector.", aliases=["super-green", "some-green"], ) camp1 = stix2.v21.Campaign(id=CAMPAIGN_ID, **CAMP_KWARGS) camp2 = stix2.v21.Campaign(id=CAMPAIGN_ID, **CAMP_KWARGS) env = stix2.Environment().object_similarity(camp1, camp2) assert round(env) == 100 def test_object_similarity_on_same_identity1(): iden1 = stix2.v21.Identity(id=IDENTITY_ID, **IDENTITY_KWARGS) iden2 = stix2.v21.Identity(id=IDENTITY_ID, **IDENTITY_KWARGS) env = stix2.Environment().object_similarity(iden1, iden2) assert round(env) == 100 def test_object_similarity_on_same_identity2(): IDEN_KWARGS = dict( name="John Smith", identity_class="individual", sectors=["government", "infrastructure"], ) iden1 = stix2.v21.Identity(id=IDENTITY_ID, **IDEN_KWARGS) iden2 = stix2.v21.Identity(id=IDENTITY_ID, **IDEN_KWARGS) env = stix2.Environment().object_similarity(iden1, iden2) assert round(env) == 100 def test_object_similarity_on_same_indicator(): ind1 = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) ind2 = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) env = stix2.Environment().object_similarity(ind1, ind2) assert round(env) == 100 def test_object_similarity_on_same_location1(): location_kwargs = dict(latitude=45, longitude=179) loc1 = stix2.v21.Location(id=LOCATION_ID, **location_kwargs) loc2 = stix2.v21.Location(id=LOCATION_ID, **location_kwargs) env = stix2.Environment().object_similarity(loc1, loc2) assert round(env) == 100 def test_object_similarity_on_same_location2(): location_kwargs = dict( latitude=38.889, longitude=-77.023, region="northern-america", country="us", ) loc1 = stix2.v21.Location(id=LOCATION_ID, **location_kwargs) loc2 = stix2.v21.Location(id=LOCATION_ID, **location_kwargs) env = stix2.Environment().object_similarity(loc1, loc2) assert round(env) == 100 def test_object_similarity_location_with_no_latlong(): loc_kwargs = dict(country="US", administrative_area="US-DC") loc1 = stix2.v21.Location(id=LOCATION_ID, **LOCATION_KWARGS) loc2 = stix2.v21.Location(id=LOCATION_ID, **loc_kwargs) env = stix2.Environment().object_similarity(loc1, loc2) assert round(env) != 100 def test_object_similarity_on_same_malware(): malw1 = stix2.v21.Malware(id=MALWARE_ID, **MALWARE_KWARGS) malw2 = stix2.v21.Malware(id=MALWARE_ID, **MALWARE_KWARGS) env = stix2.Environment().object_similarity(malw1, malw2) assert round(env) == 100 def test_object_similarity_on_same_threat_actor1(): ta1 = stix2.v21.ThreatActor(id=THREAT_ACTOR_ID, **THREAT_ACTOR_KWARGS) ta2 = stix2.v21.ThreatActor(id=THREAT_ACTOR_ID, **THREAT_ACTOR_KWARGS) env = stix2.Environment().object_similarity(ta1, ta2) assert round(env) == 100 def test_object_similarity_on_same_threat_actor2(): THREAT_KWARGS = dict( threat_actor_types=["crime-syndicate"], aliases=["super-evil"], name="Evil Org", ) ta1 = stix2.v21.ThreatActor(id=THREAT_ACTOR_ID, **THREAT_KWARGS) ta2 = stix2.v21.ThreatActor(id=THREAT_ACTOR_ID, **THREAT_KWARGS) env = stix2.Environment().object_similarity(ta1, ta2) assert round(env) == 100 def test_object_similarity_on_same_tool(): tool1 = stix2.v21.Tool(id=TOOL_ID, **TOOL_KWARGS) tool2 = stix2.v21.Tool(id=TOOL_ID, **TOOL_KWARGS) env = stix2.Environment().object_similarity(tool1, tool2) assert round(env) == 100 def test_object_similarity_on_same_vulnerability1(): vul1 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULNERABILITY_KWARGS) vul2 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULNERABILITY_KWARGS) prop_scores = {} env = stix2.Environment().object_similarity(vul1, vul2, prop_scores) assert round(env) == 100 assert round(prop_scores["matching_score"]) == 30 assert round(prop_scores["sum_weights"]) == 30 def test_object_equivalence_on_same_vulnerability1(): vul1 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULNERABILITY_KWARGS) vul2 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULNERABILITY_KWARGS) prop_scores = {} env = stix2.Environment().object_equivalence(vul1, vul2, prop_scores) assert env is True assert round(prop_scores["matching_score"]) == 30 assert round(prop_scores["sum_weights"]) == 30 def test_object_similarity_on_same_vulnerability2(): VULN_KWARGS1 = dict( name="Heartbleed", external_references=[ { "url": "https://example", "source_name": "some-source", }, ], ) VULN_KWARGS2 = dict( name="Foo", external_references=[ { "url": "https://example2", "source_name": "some-source2", }, ], ) vul1 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULN_KWARGS1) vul2 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULN_KWARGS2) prop_scores = {} env = stix2.Environment().object_similarity(vul1, vul2, prop_scores) assert round(env) == 0.0 assert round(prop_scores["matching_score"]) == 0 assert round(prop_scores["sum_weights"]) == 100 def test_object_equivalence_on_same_vulnerability2(): VULN_KWARGS1 = dict( name="Heartbleed", external_references=[ { "url": "https://example", "source_name": "some-source", }, ], ) VULN_KWARGS2 = dict( name="Foo", external_references=[ { "url": "https://example2", "source_name": "some-source2", }, ], ) vul1 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULN_KWARGS1) vul2 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULN_KWARGS2) prop_scores = {} env = stix2.Environment().object_equivalence(vul1, vul2, prop_scores) assert env is False assert round(prop_scores["matching_score"]) == 0 assert round(prop_scores["sum_weights"]) == 100 def test_object_similarity_on_unknown_object(): CUSTOM_KWARGS1 = dict( type="x-foobar", id="x-foobar--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061", name="Heartbleed", external_references=[ { "url": "https://example", "source_name": "some-source", }, ], ) CUSTOM_KWARGS2 = dict( type="x-foobar", id="x-foobar--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061", name="Foo", external_references=[ { "url": "https://example2", "source_name": "some-source2", }, ], ) def _x_foobar_checks(obj1, obj2, **weights): matching_score = 0.0 sum_weights = 0.0 if stix2.equivalence.object.check_property_present("external_references", obj1, obj2): w = weights["external_references"] sum_weights += w matching_score += w * stix2.equivalence.object.partial_external_reference_based( obj1["external_references"], obj2["external_references"], ) if stix2.equivalence.object.check_property_present("name", obj1, obj2): w = weights["name"] sum_weights += w matching_score += w * stix2.equivalence.object.partial_string_based(obj1["name"], obj2["name"]) return matching_score, sum_weights weights = { "x-foobar": { "external_references": 40, "name": 60, "method": _x_foobar_checks, }, "_internal": { "ignore_spec_version": False, }, } cust1 = stix2.parse(CUSTOM_KWARGS1, allow_custom=True) cust2 = stix2.parse(CUSTOM_KWARGS2, allow_custom=True) env = stix2.Environment().object_similarity(cust1, cust2, **weights) assert round(env) == 0 def test_object_similarity_different_type_raises(): with pytest.raises(ValueError) as excinfo: vul1 = stix2.v21.Vulnerability(id=VULNERABILITY_ID, **VULNERABILITY_KWARGS) ind1 = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) stix2.Environment().object_similarity(vul1, ind1) assert str(excinfo.value) == "The objects to compare must be of the same type!" def test_object_similarity_different_spec_version_raises(): with pytest.raises(ValueError) as excinfo: V20_KWARGS = dict( labels=['malicious-activity'], pattern="[file:hashes.MD5 = 'd41d8cd98f00b204e9800998ecf8427e']", ) ind1 = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) ind2 = stix2.v20.Indicator(id=INDICATOR_ID, **V20_KWARGS) stix2.Environment().object_similarity(ind1, ind2) assert str(excinfo.value) == "The objects to compare must be of the same spec version!" def test_object_similarity_zero_match(): IND_KWARGS = dict( indicator_types=["anomalous-activity"], pattern="[ipv4-addr:value = '192.168.1.1']", pattern_type="stix", valid_from="2019-01-01T12:34:56Z", labels=["APTX", "foo"], ) weights = { "indicator": { "indicator_types": (15, stix2.equivalence.object.partial_list_based), "pattern": (80, stix2.equivalence.object.custom_pattern_based), "valid_from": (5, stix2.equivalence.object.partial_timestamp_based), "tdelta": 1, # One day interval }, "_internal": { "ignore_spec_version": False, }, } ind1 = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) ind2 = stix2.v21.Indicator(id=INDICATOR_ID, **IND_KWARGS) env = stix2.Environment().object_similarity(ind1, ind2, **weights) assert round(env) == 0 env = stix2.Environment().object_similarity(ind2, ind1, **weights) assert round(env) == 0 def test_object_similarity_different_spec_version(): IND_KWARGS = dict( labels=["malicious-activity"], pattern="[ipv4-addr:value = '192.168.1.1']", ) weights = { "indicator": { "indicator_types": (15, stix2.equivalence.object.partial_list_based), "pattern": (80, stix2.equivalence.object.custom_pattern_based), "valid_from": (5, stix2.equivalence.object.partial_timestamp_based), "tdelta": 1, # One day interval }, } ind1 = stix2.v21.Indicator(id=INDICATOR_ID, **INDICATOR_KWARGS) ind2 = stix2.v20.Indicator(id=INDICATOR_ID, **IND_KWARGS) env = stix2.Environment().object_similarity(ind1, ind2, ignore_spec_version=True, **weights) assert round(env) == 0 env = stix2.Environment().object_similarity(ind2, ind1, ignore_spec_version=True, **weights) assert round(env) == 0 @pytest.mark.parametrize( "refs1,refs2,ret_val", [ ( [ { "url": "https://attack.mitre.org/techniques/T1150", "source_name": "mitre-attack", "external_id": "T1150", }, { "url": "https://researchcenter.paloaltonetworks.com/2016/09/unit42-sofacys-komplex-os-x-trojan/", "source_name": "Sofacy Komplex Trojan", "description": "Dani Creus, Tyler Halfpop, Robert Falcone. (2016, September 26). Sofacy's 'Komplex' OS X Trojan. Retrieved July 8, 2017.", }, ], [ { "url": "https://attack.mitre.org/techniques/T1129", "source_name": "mitre-attack", "external_id": "T1129", }, { "url": "https://en.wikipedia.org/wiki/Microsoft_Windows_library_files", "source_name": "Wikipedia Windows Library Files", "description": "Wikipedia. (2017, January 31). Microsoft Windows library files. Retrieved February 13, 2017.", }, ], 0.0, ), ( [ { "url": "https://attack.mitre.org/techniques/T1129", "source_name": "mitre-attack", "external_id": "T1129", }, ], [ { "url": "https://attack.mitre.org/techniques/T1129", "source_name": "mitre-attack", "external_id": "T1129", }, { "url": "https://en.wikipedia.org/wiki/Microsoft_Windows_library_files", "source_name": "Wikipedia Windows Library Files", "description": "Wikipedia. (2017, January 31). Microsoft Windows library files. Retrieved February 13, 2017.", }, ], 1.0, ), ( [ { "url": "https://example", "source_name": "some-source", }, ], [ { "url": "https://example", "source_name": "some-source", }, ], 1.0, ), ], ) def test_object_similarity_external_references(refs1, refs2, ret_val): value = stix2.equivalence.object.partial_external_reference_based(refs1, refs2) assert value == ret_val def test_object_similarity_timestamp(): t1 = "2018-10-17T00:14:20.652Z" t2 = "2018-10-17T12:14:20.652Z" assert stix2.equivalence.object.partial_timestamp_based(t1, t2, 1) == 0.5 def test_object_similarity_exact_match(): t1 = "2018-10-17T00:14:20.652Z" t2 = "2018-10-17T12:14:20.652Z" assert stix2.equivalence.object.exact_match(t1, t2) == 0.0 def test_no_datastore_fallsback_list_based_check_for_refs_check(): r1 = stix2.v21.Report(id=REPORT_ID, **REPORT_KWARGS) r2 = stix2.v21.Report(id=REPORT_ID, **REPORT_KWARGS) prop_scores = {} assert stix2.Environment().object_similarity(r1, r2, prop_scores) == 100.0 assert prop_scores["object_refs"]["check_type"] == "partial_list_based" def custom_semantic_equivalence_method(obj1, obj2, **weights): return 96.0, 100.0 def test_object_similarity_method_provided(): # Because `method` is provided, `partial_list_based` will be ignored TOOL2_KWARGS = dict( name="Random Software", tool_types=["information-gathering"], ) weights = { "tool": { "tool_types": (20, stix2.equivalence.object.partial_list_based), "name": (80, stix2.equivalence.object.partial_string_based), "method": custom_semantic_equivalence_method, }, } tool1 = stix2.v21.Tool(id=TOOL_ID, **TOOL_KWARGS) tool2 = stix2.v21.Tool(id=TOOL_ID, **TOOL2_KWARGS) env = stix2.Environment().object_similarity(tool1, tool2, **weights) assert round(env) == 96 def test_object_similarity_prop_scores(): TOOL2_KWARGS = dict( name="Random Software", tool_types=["information-gathering"], ) prop_scores = {} tool1 = stix2.v21.Tool(id=TOOL_ID, **TOOL_KWARGS) tool2 = stix2.v21.Tool(id=TOOL_ID, **TOOL2_KWARGS) stix2.Environment().object_similarity(tool1, tool2, prop_scores) assert len(prop_scores) == 4 assert round(prop_scores["matching_score"], 1) == 8.9 assert round(prop_scores["sum_weights"], 1) == 100.0 def custom_semantic_equivalence_method_prop_scores(obj1, obj2, prop_scores, **weights): prop_scores["matching_score"] = 96.0 prop_scores["sum_weights"] = 100.0 return 96.0, 100.0 def test_object_similarity_prop_scores_method_provided(): TOOL2_KWARGS = dict( name="Random Software", tool_types=["information-gathering"], ) weights = { "tool": { "tool_types": 20, "name": 80, "method": custom_semantic_equivalence_method_prop_scores, }, } prop_scores = {} tool1 = stix2.v21.Tool(id=TOOL_ID, **TOOL_KWARGS) tool2 = stix2.v21.Tool(id=TOOL_ID, **TOOL2_KWARGS) env = stix2.Environment().object_similarity(tool1, tool2, prop_scores, **weights) assert round(env) == 96 assert len(prop_scores) == 2 assert prop_scores["matching_score"] == 96.0 assert prop_scores["sum_weights"] == 100.0 def test_versioned_checks(ds, ds2): # Testing internal method weights = stix2.equivalence.graph.WEIGHTS.copy() weights.update({ "_internal": { "ignore_spec_version": True, "versioning_checks": True, "max_depth": 1, }, }) score = stix2.equivalence.object._versioned_checks(INDICATOR_ID, INDICATOR_ID, ds, ds2, **weights) assert round(score) == 100 def test_semantic_check_with_versioning(ds, ds2): weights = stix2.equivalence.graph.WEIGHTS.copy() weights.update({ "_internal": { "ignore_spec_version": False, "versioning_checks": True, "ds1": ds, "ds2": ds2, "max_depth": 1, }, }) ind = stix2.v21.Indicator( **dict( indicator_types=["malicious-activity"], pattern_type="stix", pattern="[file:hashes.'SHA-256' = 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855']", valid_from="2017-01-01T12:34:56Z", external_references=[ { "source_name": "unknown", "url": "https://examplewebsite2.com/", }, ], object_marking_refs=[stix2.v21.TLP_WHITE], ) ) ds.add(ind) score = stix2.equivalence.object.reference_check(ind.id, INDICATOR_ID, ds, ds2, **weights) assert round(score) == 0 # Since pattern is different score is really low def test_list_semantic_check(ds, ds2): weights = stix2.equivalence.graph.WEIGHTS.copy() weights.update({ "_internal": { "ignore_spec_version": False, "versioning_checks": False, "ds1": ds, "ds2": ds2, "max_depth": 1, }, }) object_refs1 = [ "malware--9c4638ec-f1de-4ddb-abf4-1b760417654e", "relationship--06520621-5352-4e6a-b976-e8fa3d437ffd", "indicator--a740531e-63ff-4e49-a9e1-a0a3eed0e3e7", ] object_refs2 = [ "campaign--8e2e2d2b-17d4-4cbf-938f-98ee46b3cd3f", "identity--311b2d2d-f010-4473-83ec-1edf84858f4c", "indicator--a740531e-63ff-4e49-a9e1-a0a3eed0e3e7", "malware--9c4638ec-f1de-4ddb-abf4-1b760417654e", "malware--9c4638ec-f1de-4ddb-abf4-1b760417654e", "relationship--06520621-5352-4e6a-b976-e8fa3d437ffd", "relationship--181c9c09-43e6-45dd-9374-3bec192f05ef", "relationship--a0cbb21c-8daf-4a7f-96aa-7155a4ef8f70", ] score = stix2.equivalence.object.list_reference_check( object_refs1, object_refs2, ds, ds2, **weights, ) assert round(score) == 1 score = stix2.equivalence.object.list_reference_check( object_refs2, object_refs1, ds2, ds, **weights, ) assert round(score) == 1 def test_graph_similarity_raises_value_error(ds): with pytest.raises(ValueError): prop_scores1 = {} stix2.Environment().graph_similarity(ds, ds2, prop_scores1, max_depth=-1) def test_graph_similarity_with_filesystem_source(ds, fs): prop_scores1 = {} env1 = stix2.Environment().graph_similarity( fs, ds, prop_scores1, ignore_spec_version=True, versioning_checks=False, max_depth=1, ) # Switching parameters prop_scores2 = {} env2 = stix2.Environment().graph_similarity( ds, fs, prop_scores2, ignore_spec_version=True, versioning_checks=False, max_depth=1, ) assert round(env1) == 23 assert round(prop_scores1["matching_score"]) == 411 assert round(prop_scores1["len_pairs"]) == 18 assert round(env2) == 23 assert round(prop_scores2["matching_score"]) == 411 assert round(prop_scores2["len_pairs"]) == 18 prop_scores1["matching_score"] = round(prop_scores1["matching_score"], 3) prop_scores2["matching_score"] = round(prop_scores2["matching_score"], 3) assert json.dumps(prop_scores1, sort_keys=True, indent=4) == json.dumps(prop_scores2, sort_keys=True, indent=4) def test_depth_limiting(): g1 = [ { "type": "foo", "id": "foo--07f9dd2a-1cce-45bb-8cbe-dba3f007aafd", "spec_version": "2.1", "created": "1986-02-08T00:20:17Z", "modified": "1989-12-11T06:54:29Z", "some1_ref": "foo--700a8a3c-9936-412f-b4eb-ede466476180", "some2_ref": "foo--f4a999a3-df94-499d-9cac-6c02e21775ee", }, { "type": "foo", "id": "foo--700a8a3c-9936-412f-b4eb-ede466476180", "spec_version": "2.1", "created": "1989-01-06T10:31:54Z", "modified": "1995-06-18T10:25:01Z", "some1_ref": "foo--705afd45-eb56-43fc-a214-313d63d199a3", }, { "type": "foo", "id": "foo--705afd45-eb56-43fc-a214-313d63d199a3", "spec_version": "2.1", "created": "1977-11-06T21:19:29Z", "modified": "1997-12-02T20:33:34Z", }, { "type": "foo", "id": "foo--f4a999a3-df94-499d-9cac-6c02e21775ee", "spec_version": "2.1", "created": "1991-09-17T00:40:52Z", "modified": "1992-12-06T11:02:47Z", "name": "alice", }, ] g2 = [ { "type": "foo", "id": "foo--71570479-3e6e-48d2-81fb-897454dec55d", "spec_version": "2.1", "created": "1975-12-22T05:20:38Z", "modified": "1980-11-11T01:09:03Z", "some1_ref": "foo--4aeda39b-31fa-4ffb-a847-d8edc175a579", "some2_ref": "foo--941e48d6-3100-4419-9e8c-cf1eb59e71b2", }, { "type": "foo", "id": "foo--4aeda39b-31fa-4ffb-a847-d8edc175a579", "spec_version": "2.1", "created": "1976-01-05T08:32:03Z", "modified": "1980-11-09T05:41:02Z", "some1_ref": "foo--689252c3-5d20-43ff-bbf7-c8e45d713768", }, { "type": "foo", "id": "foo--689252c3-5d20-43ff-bbf7-c8e45d713768", "spec_version": "2.1", "created": "1974-09-11T18:56:30Z", "modified": "1976-10-31T11:59:43Z", }, { "type": "foo", "id": "foo--941e48d6-3100-4419-9e8c-cf1eb59e71b2", "spec_version": "2.1", "created": "1985-01-03T01:07:03Z", "modified": "1992-07-20T21:32:31Z", "name": "alice", }, ] mem_store1 = stix2.MemorySource(g1) mem_store2 = stix2.MemorySource(g2) custom_weights = { "foo": { "some1_ref": (33, stix2.equivalence.object.reference_check), "some2_ref": (33, stix2.equivalence.object.reference_check), "name": (34, stix2.equivalence.object.partial_string_based), }, } prop_scores1 = {} env1 = stix2.equivalence.graph.graph_similarity( mem_store1, mem_store2, prop_scores1, **custom_weights ) assert round(env1) == 38 assert round(prop_scores1["matching_score"]) == 300 assert round(prop_scores1["len_pairs"]) == 8 # from 'alice' check in de-reference assert prop_scores1['summary']['foo--71570479-3e6e-48d2-81fb-897454dec55d']['prop_score']['some2_ref']['weight'] == 33 assert prop_scores1['summary']['foo--07f9dd2a-1cce-45bb-8cbe-dba3f007aafd']['prop_score']['some2_ref']['weight'] == 33 # Switching parameters prop_scores2 = {} env2 = stix2.equivalence.graph.graph_similarity( mem_store2, mem_store1, prop_scores2, **custom_weights ) assert round(env2) == 38 assert round(prop_scores2["matching_score"]) == 300 assert round(prop_scores2["len_pairs"]) == 8 # from 'alice' check in de-reference assert prop_scores2['summary']['foo--71570479-3e6e-48d2-81fb-897454dec55d']['prop_score']['some2_ref']['weight'] == 33 assert prop_scores2['summary']['foo--07f9dd2a-1cce-45bb-8cbe-dba3f007aafd']['prop_score']['some2_ref']['weight'] == 33 def test_graph_similarity_with_duplicate_graph(ds): prop_scores = {} env = stix2.Environment().graph_similarity(ds, ds, prop_scores) assert round(env) == 100 assert round(prop_scores["matching_score"]) == 800 assert round(prop_scores["len_pairs"]) == 8 def test_graph_similarity_with_versioning_check_on(ds2, ds): prop_scores1 = {} env1 = stix2.Environment().graph_similarity(ds, ds2, prop_scores1, versioning_checks=True) assert round(env1) == 88 assert round(prop_scores1["matching_score"]) == 789 assert round(prop_scores1["len_pairs"]) == 9 # Switching parameters prop_scores2 = {} env2 = stix2.Environment().graph_similarity(ds2, ds, prop_scores2, versioning_checks=True) assert round(env2) == 88 assert round(prop_scores2["matching_score"]) == 789 assert round(prop_scores2["len_pairs"]) == 9 prop_scores1["matching_score"] = round(prop_scores1["matching_score"], 3) prop_scores2["matching_score"] = round(prop_scores2["matching_score"], 3) assert json.dumps(prop_scores1, sort_keys=True, indent=4) == json.dumps(prop_scores2, sort_keys=True, indent=4) def test_graph_similarity_with_versioning_check_off(ds2, ds): prop_scores1 = {} env1 = stix2.Environment().graph_similarity(ds, ds2, prop_scores1) assert round(env1) == 88 assert round(prop_scores1["matching_score"]) == 789 assert round(prop_scores1["len_pairs"]) == 9 # Switching parameters prop_scores2 = {} env2 = stix2.Environment().graph_similarity(ds2, ds, prop_scores2) assert round(env2) == 88 assert round(prop_scores2["matching_score"]) == 789 assert round(prop_scores2["len_pairs"]) == 9 prop_scores1["matching_score"] = round(prop_scores1["matching_score"], 3) prop_scores2["matching_score"] = round(prop_scores2["matching_score"], 3) assert json.dumps(prop_scores1, sort_keys=True, indent=4) == json.dumps(prop_scores2, sort_keys=True, indent=4) def test_graph_equivalence_with_filesystem_source(ds, fs): prop_scores1 = {} env1 = stix2.Environment().graph_equivalence(fs, ds, prop_scores1, ignore_spec_version=True) # Switching parameters prop_scores2 = {} env2 = stix2.Environment().graph_equivalence(ds, fs, prop_scores2, ignore_spec_version=True) assert env1 is False assert round(prop_scores1["matching_score"]) == 411 assert round(prop_scores1["len_pairs"]) == 18 assert env2 is False assert round(prop_scores2["matching_score"]) == 411 assert round(prop_scores2["len_pairs"]) == 18 prop_scores1["matching_score"] = round(prop_scores1["matching_score"], 3) prop_scores2["matching_score"] = round(prop_scores2["matching_score"], 3) assert json.dumps(prop_scores1, sort_keys=True, indent=4) == json.dumps(prop_scores2, sort_keys=True, indent=4) def test_graph_equivalence_with_duplicate_graph(ds): prop_scores = {} env = stix2.Environment().graph_equivalence(ds, ds, prop_scores) assert env is True assert round(prop_scores["matching_score"]) == 800 assert round(prop_scores["len_pairs"]) == 8 def test_graph_equivalence_with_versioning_check_on(ds2, ds): prop_scores1 = {} env1 = stix2.Environment().graph_equivalence(ds, ds2, prop_scores1, versioning_checks=True) # Switching parameters prop_scores2 = {} env2 = stix2.Environment().graph_equivalence(ds2, ds, prop_scores2, versioning_checks=True) assert env1 is True assert round(prop_scores1["matching_score"]) == 789 assert round(prop_scores1["len_pairs"]) == 9 assert env2 is True assert round(prop_scores2["matching_score"]) == 789 assert round(prop_scores2["len_pairs"]) == 9 prop_scores1["matching_score"] = round(prop_scores1["matching_score"], 3) prop_scores2["matching_score"] = round(prop_scores2["matching_score"], 3) assert json.dumps(prop_scores1, sort_keys=True, indent=4) == json.dumps(prop_scores2, sort_keys=True, indent=4) def test_graph_equivalence_with_versioning_check_off(ds2, ds): prop_scores1 = {} env1 = stix2.Environment().graph_equivalence(ds, ds2, prop_scores1) # Switching parameters prop_scores2 = {} env2 = stix2.Environment().graph_equivalence(ds2, ds, prop_scores2) assert env1 is True assert round(prop_scores1["matching_score"]) == 789 assert round(prop_scores1["len_pairs"]) == 9 assert env2 is True assert round(prop_scores2["matching_score"]) == 789 assert round(prop_scores2["len_pairs"]) == 9 prop_scores1["matching_score"] = round(prop_scores1["matching_score"], 3) prop_scores2["matching_score"] = round(prop_scores2["matching_score"], 3) assert json.dumps(prop_scores1, sort_keys=True, indent=4) == json.dumps(prop_scores2, sort_keys=True, indent=4)