diff --git a/external-import/recorded-future/src/rflib/rf_to_stix2.py b/external-import/recorded-future/src/rflib/rf_to_stix2.py index ec32762510d..32cd814b632 100644 --- a/external-import/recorded-future/src/rflib/rf_to_stix2.py +++ b/external-import/recorded-future/src/rflib/rf_to_stix2.py @@ -63,18 +63,33 @@ def __init__( self.last_seen = last_seen def to_stix_objects(self): - """Returns a list of STIX objects""" + """ + Returns the STIX objects representing this entity. + """ if not self.stix_obj: self.create_stix_objects() - return [self.author, self.stix_obj] + return [self.stix_obj] def create_stix_objects(self): """Creates STIX objects from object attributes""" pass + def _bundle_objects(self, objects): + """ + Returns bundle-ready objects: the given objects plus the author identity + they reference through 'created_by_ref', deduplicated by id. + """ + bundle_objects = {} + for stix_object in [*objects, self.author]: + if stix_object is not None: + bundle_objects[stix_object["id"]] = stix_object + return list(bundle_objects.values()) + def to_stix_bundle(self): """Returns STIX objects as a Bundle""" - return stix2.Bundle(objects=self.to_stix_objects(), allow_custom=True) + return stix2.Bundle( + objects=self._bundle_objects(self.to_stix_objects()), allow_custom=True + ) def to_json_bundle(self): """Returns STIX Bundle as JSON""" @@ -193,7 +208,9 @@ def _create_rel(self, relationship_type, target_id): def to_stix_bundle(self): """Returns STIX objects as a Bundle""" return stix2.Bundle( - objects=self.objects if self.objects else self.to_stix_objects(), + objects=self._bundle_objects( + self.objects if self.objects else self.to_stix_objects() + ), allow_custom=True, ) @@ -308,7 +325,9 @@ def _create_obs(self): def to_stix_bundle(self): """Returns STIX objects as a Bundle""" - return stix2.Bundle(objects=self.objects, allow_custom=True) + return stix2.Bundle( + objects=self._bundle_objects(self.objects), allow_custom=True + ) class Domain(Indicator): @@ -480,7 +499,9 @@ def create_stix_objects(self): def to_stix_bundle(self): """Returns STIX objects as a Bundle""" return stix2.Bundle( - objects=self.objects if self.objects else self.to_stix_objects(), + objects=self._bundle_objects( + self.objects if self.objects else self.to_stix_objects() + ), allow_custom=True, ) @@ -620,7 +641,9 @@ def create_stix_objects(self): def to_stix_bundle(self): """Returns STIX objects as a Bundle""" return stix2.Bundle( - objects=self.objects if self.objects else self.to_stix_objects(), + objects=self._bundle_objects( + self.objects if self.objects else self.to_stix_objects() + ), allow_custom=True, ) @@ -906,7 +929,9 @@ def _create_rel(self, relationship_type, source_id): def to_stix_bundle(self): """Returns STIX objects as a Bundle""" return stix2.Bundle( - objects=self.objects if self.objects else self.to_stix_objects(), + objects=self._bundle_objects( + self.objects if self.objects else self.to_stix_objects() + ), allow_custom=True, ) @@ -922,7 +947,6 @@ def __init__(self, name, _type, content, author, tlp=None, first_seen=None): self.type = _type self.content = content self.stix_obj = None - self.author = author if self.type not in ("yara", "snort", "sigma"): msg = f"[ANALYST NOTES] Detection rule of type {self.type} is not supported" @@ -1130,6 +1154,7 @@ class StixNote: "Validated Intelligence Event": "Observed-Data", "Weekly Threat Landscape": "Threat-Report", "YARA Rule": "Indicator", + "Vulnerability Intelligence": "Vulnerability", } def __init__( @@ -1520,7 +1545,9 @@ def _create_report_types(self, topics): def to_stix_objects(self): """Returns a list of STIX objects""" - report_object_refs = [obj.id for obj in self.objects] + # 'self.objects' only holds the note content: the author identity is added + # to the bundle below, but is never referenced in the Report 'object_refs' + report_object_refs = list(dict.fromkeys(obj.id for obj in self.objects)) # Report in STIX lib must have at least one object_refs even if there is no object_refs # Use a subclass of Report to make the object_refs optional diff --git a/external-import/recorded-future/tests/tests_connector/test_rf_to_stix2.py b/external-import/recorded-future/tests/tests_connector/test_rf_to_stix2.py index 31ac290caec..ff7deb98bdc 100644 --- a/external-import/recorded-future/tests/tests_connector/test_rf_to_stix2.py +++ b/external-import/recorded-future/tests/tests_connector/test_rf_to_stix2.py @@ -1,9 +1,11 @@ import json +from unittest.mock import MagicMock import pytest from pycti import Identity as PyctiIdentity from rflib.rf_to_stix2 import ENTITY_TYPE_MAPPER from rflib.rf_to_stix2 import IPAddress as RFIPAddress +from rflib.rf_to_stix2 import StixNote from rflib.rf_to_stix2 import Vulnerability as RFVulnerability from stix2 import ( URL, @@ -41,19 +43,19 @@ "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", [Indicator, File, Relationship], ), - ("MitreAttackIdentifier", "test mitreattack", [Identity, AttackPattern]), - ("Company", "test company", [Identity, Identity]), - ("Person", "test person", [Identity, Identity]), - ("Organization", "test organization", [Identity, Identity]), - ("Malware", "test malware", [Identity, Malware]), + ("MitreAttackIdentifier", "test mitreattack", [AttackPattern]), + ("Company", "test company", [Identity]), + ("Person", "test person", [Identity]), + ("Organization", "test organization", [Identity]), + ("Malware", "test malware", [Malware]), ("CyberVulnerability", "test cybervuln", [Vulnerability]), ("Product", "test product", [Software]), ("Country", "test country", [Location]), ("City", "test city", [Location]), ("ProvinceOrState", "test province", [Location]), - ("Industry", "test industry", [Identity, Identity]), + ("Industry", "test industry", [Identity]), ("Operation", "test operation", [Campaign]), - ("Threat Actor", "test threat actor", [Identity, ThreatActor]), + ("Threat Actor", "test threat actor", [ThreatActor]), ], ) # Scenario: Each Recorded Future entity type maps to the expected STIX2 object type(s) @@ -223,6 +225,76 @@ def test_indicator_map_data_respects_risk_list_disabled(): _then_contains_no_intrusion_set(indicator.related_entities) +# Scenario: The author identity is never part of an entity's own STIX objects +@pytest.mark.parametrize( + "rf_type, name", + [ + ("MitreAttackIdentifier", "test mitreattack"), + ("Company", "test company"), + ("Person", "test person"), + ("Organization", "test organization"), + ("Malware", "test malware"), + ("Industry", "test industry"), + ("Threat Actor", "test threat actor"), + ], +) +def test_entity_stix_objects_exclude_the_author(rf_type, name): + # Given a valid author identity and TLP marking + author = _given_author() + tlp = _given_tlp() + # And an RF entity of type with name + rf_object = _given_rf_entity(rf_type, name, author, tlp) + + # When the entity is converted to STIX objects + stix_objects = _when_to_stix_objects(rf_object) + + # Then the author is not part of the entity's own objects + _then_objects_exclude(stix_objects, author) + + +# Scenario: An entity bundle still carries the author it references (risk lists, threat maps) +@pytest.mark.parametrize( + "rf_type, name", + [ + ("IpAddress", "3.3.3.3"), + ("Malware", "test malware"), + ("CyberVulnerability", "CVE-2024-1111"), + ], +) +def test_entity_bundle_contains_the_author_exactly_once(rf_type, name): + # Given a valid author identity and TLP marking + author = _given_author() + tlp = _given_tlp() + # And an RF entity of type whose bundle has been built + rf_object = _given_rf_entity(rf_type, name, author, tlp) + + # When the entity is converted to a STIX bundle + bundle = _when_built_to_stix_bundle(rf_object) + + # Then the author identity referenced by 'created_by_ref' is in the bundle, once + _then_objects_contain_once(bundle.objects, author) + + +# Scenario: The report creator is not listed among the report's entities (issue #7003) +def test_analyst_note_report_does_not_reference_the_author(): + # Given an analyst note holding a Malware and a Company entity + note = _given_stix_note() + _when_note_converted_from_json(note, _given_analyst_note_json()) + + # When the note is converted to STIX objects + stix_objects = note.to_stix_objects() + report = _then_single_report(stix_objects) + + # Then the author is the report creator + assert report.created_by_ref == note.author.id + # And the author is not referenced as one of the report's entities + assert note.author.id not in report.object_refs + # And the author identity is still part of the bundle, once + _then_objects_contain_once(stix_objects, note.author) + # And the note entities are referenced by the report + assert len(report.object_refs) == 2 + + # ── Given helpers ──────────────────────────────────────────────────────────── @@ -250,6 +322,27 @@ def _given_ip_indicator(ip, author, tlp): return RFIPAddress(ip, "IpAddress", author, tlp) +def _given_stix_note(): + return StixNote(opencti_helper=MagicMock(), tas=[], rfapi=MagicMock()) + + +def _given_analyst_note_json(): + return { + "id": "note-id", + "attributes": { + "title": "Test analyst note", + "text": "Some intelligence content", + "published": "2026-08-20T00:00:00.000Z", + "topic": [{"name": "Flash Report"}], + "attachments": [], + "note_entities": [ + {"id": "entity-1", "type": "Malware", "name": "Test malware"}, + {"id": "entity-2", "type": "Company", "name": "Test company"}, + ], + }, + } + + def _given_vuln_risk_row(risk, threat_actor_name): links = [ { @@ -297,6 +390,15 @@ def _when_to_stix_objects(rf_object): return rf_object.to_stix_objects() +def _when_built_to_stix_bundle(rf_object): + rf_object.build_bundle(rf_object) + return rf_object.to_stix_bundle() + + +def _when_note_converted_from_json(note, note_json): + note.from_json(note_json, _given_tlp()) + + def _when_vuln_map_data(vuln, rf_row, tlp, related_entity_types): vuln.map_data(rf_row, tlp, risklist_related_entities=related_entity_types) @@ -331,10 +433,29 @@ def _when_indicator_map_data_with_ta_scope( def _then_stix_types_match(stix_objects, expected_types): + assert len(stix_objects) == len( + expected_types + ), f"Expected {len(expected_types)} STIX objects, got {len(stix_objects)}" for i, stix_obj in enumerate(stix_objects): assert isinstance(stix_obj, expected_types[i]) +def _then_objects_exclude(stix_objects, excluded_object): + found = [obj for obj in stix_objects if obj.id == excluded_object.id] + assert len(found) == 0, f"Expected {excluded_object.id} to be absent, got {found}" + + +def _then_single_report(stix_objects): + reports = [obj for obj in stix_objects if obj["type"] == "report"] + assert len(reports) == 1, f"Expected exactly 1 report, got {len(reports)}" + return reports[0] + + +def _then_objects_contain_once(stix_objects, expected_object): + found = [obj for obj in stix_objects if obj.id == expected_object.id] + assert len(found) == 1, f"Expected exactly 1 {expected_object.id}, got {len(found)}" + + def _then_contains_intrusion_set(related_entities, expected_count=1): found = [e for e in related_entities if isinstance(e, IntrusionSet)] assert (