feat(google-ti-feeds): add malware/threat actor filter (#7024)

external-import/google-ti-feeds/.env.sample

@@ -1,14 +1,16 @@
 OPENCTI_URL='http://localhost:PORT'
 OPENCTI_TOKEN='ChangeMe'
 CONNECTOR_ID='ChangeMe'
 GTI_API_KEY='ChangeMe'
 
 # GTI_IMPORT_INDICATORS=false
 # GTI_INDICATOR_TYPES=file,ip,url,domain
 # GTI_INDICATOR_IMPORT_START_DATE=PT1H
 # GTI_INDICATOR_MIN_SCORE=50
+# GTI_INDICATOR_REQUIRE_MALWARE_FAMILY=false
+# GTI_INDICATOR_REQUIRE_THREAT_ACTOR=false
 
 # GTI_IMPORT_SOFTWARE_TOOLKITS=false
 # GTI_SOFTWARE_TOOLKIT_IMPORT_START_DATE=P1D
 # GTI_SOFTWARE_TOOLKIT_ORIGINS=google threat intelligence
 # GTI_SOFTWARE_TOOLKIT_EXTRA_FILTERS=

external-import/google-ti-feeds/README.md

@@ -298,6 +298,8 @@ Unlike the other collections which use the `/collections` search endpoint, the I
 |-----------------------------------|-----------------------------------------------------------------------------------|-----------------------------------|------------------------------------------------------------------------------------------------------------------------------|
 | `GTI_INDICATOR_TYPES`             | List of IOC types to import                                                       | `["file", "ip", "url", "domain"]` | `["file", "ip", "url", "domain"]`                                                                                            |
 | `GTI_INDICATOR_MIN_SCORE`         | Minimum GTI score an indicator must have to be imported via Delta Sync            | `50`                               | Integer between `0` and `100`, or unset/`100` to disable the filter                                                          |
+| `GTI_INDICATOR_REQUIRE_MALWARE_FAMILY` | Only import indicators associated with at least one Malware Family           | `false`                            | `true` / `false`                                                                                                             |
+| `GTI_INDICATOR_REQUIRE_THREAT_ACTOR`   | Only import indicators associated with at least one Threat Actor             | `false`                            | `true` / `false`                                                                                                             |
 
 ---
 

external-import/google-ti-feeds/__metadata__/CONNECTOR_CONFIG_DOC.md

@@ -26,6 +26,8 @@ Below is an exhaustive enumeration of all configurable parameters available, eac
 | GTI_INDICATOR_TYPES | `array` |  | string | `["file", "ip", "url", "domain"]` | Comma-separated list of IOC types to import. Allowed: file, ip, url, domain |  |
 | GTI_INDICATOR_IMPORT_START_DATE | `string` |  | Format: [`duration`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | `"P1D"` | ISO 8601 duration string specifying how far back to import indicators on first run (e.g. P1D for 1 day, P7D for 7 days). Must be greater than 1 hour (the IOC delta package granularity). |  |
 | GTI_INDICATOR_MIN_SCORE | `integer` |  | `0 <= x <= 100` | `50` | Minimum GTI score (0-100) an indicator must have to be imported via the Steady-State IOC Deltas API. Indicators with a lower score are discarded. Indicators without a score are always imported. Set to 100 or leave unset (None) to disable the filter entirely. |  |
+| GTI_INDICATOR_REQUIRE_MALWARE_FAMILY | `boolean` |  | boolean | `false` | Only import indicators that have at least one association with a Malware Family. Combined with GTI_INDICATOR_REQUIRE_THREAT_ACTOR using OR logic, and with GTI_INDICATOR_MIN_SCORE using AND logic. |  |
+| GTI_INDICATOR_REQUIRE_THREAT_ACTOR | `boolean` |  | boolean | `false` | Only import indicators that have at least one association with a Threat Actor. Combined with GTI_INDICATOR_REQUIRE_MALWARE_FAMILY using OR logic, and with GTI_INDICATOR_MIN_SCORE using AND logic. |  |
 | GTI_CAMPAIGN_IMPORT_START_DATE | `string` |  | Format: [`duration`](https://json-schema.org/understanding-json-schema/reference/string#built-in-formats) | `"P1D"` | ISO 8601 duration string specifying how far back to import campaigns (e.g., P1D for 1 day, P7D for 7 days) |  |
 | GTI_IMPORT_CAMPAIGNS | `boolean` |  | boolean | `false` | Whether to enable importing campaign data from GTI |  |
 | GTI_CAMPAIGN_ORIGINS | `array` |  | string | `["google threat intelligence"]` | Comma-separated list of campaign origins to import, or 'All' for all origins. Allowed values: All, partner, crowdsourced, google threat intelligence | ```All```, ```partner```, ```google threat intelligence```, ```crowdsourced``` |

external-import/google-ti-feeds/__metadata__/connector_config_schema.json

@@ -173,6 +173,16 @@
       "minimum": 0,
       "type": "integer"
     },
+    "GTI_INDICATOR_REQUIRE_MALWARE_FAMILY": {
+      "default": false,
+      "description": "Only import indicators that have at least one association with a Malware Family. Combined with GTI_INDICATOR_REQUIRE_THREAT_ACTOR using OR logic, and with GTI_INDICATOR_MIN_SCORE using AND logic.",
+      "type": "boolean"
+    },
+    "GTI_INDICATOR_REQUIRE_THREAT_ACTOR": {
+      "default": false,
+      "description": "Only import indicators that have at least one association with a Threat Actor. Combined with GTI_INDICATOR_REQUIRE_MALWARE_FAMILY using OR logic, and with GTI_INDICATOR_MIN_SCORE using AND logic.",
+      "type": "boolean"
+    },
     "GTI_CAMPAIGN_IMPORT_START_DATE": {
       "default": "P1D",
       "description": "ISO 8601 duration string specifying how far back to import campaigns (e.g., P1D for 1 day, P7D for 7 days)",

external-import/google-ti-feeds/config.yml.sample

@@ -11,3 +11,6 @@ gti:
   # import_indicators: false
   # indicator_types: file,ip,url,domain
   # indicator_import_start_date: PT1H
+  # indicator_min_score: 50
+  # indicator_require_malware_family: false
+  # indicator_require_threat_actor: false

external-import/google-ti-feeds/connector/src/custom/configs/indicator/gti_config_indicator.py

@@ -40,6 +40,18 @@ class GTIIndicatorConfig(GTIBaseConfig):
         "score are discarded. Indicators without a score are always imported. "
         "Set to 100 or leave unset (None) to disable the filter entirely.",
     )
+    indicator_require_malware_family: bool = Field(
+        default=False,
+        description="Only import indicators that have at least one association "
+        "with a Malware Family. Combined with indicator_require_threat_actor "
+        "using OR logic, and with indicator_min_score using AND logic.",
+    )
+    indicator_require_threat_actor: bool = Field(
+        default=False,
+        description="Only import indicators that have at least one association "
+        "with a Threat Actor. Combined with indicator_require_malware_family "
+        "using OR logic, and with indicator_min_score using AND logic.",
+    )
 
     @field_validator("indicator_import_start_date", mode="after")
     @classmethod

external-import/google-ti-feeds/connector/src/custom/convert_to_stix/indicator/convert_to_stix_indicator.py

@@ -74,6 +74,41 @@ def _is_below_min_score(self, entry: IOCDeltaEntry) -> bool:
 
         return attrs.gti_assessment.threat_score.value < min_score
 
+    def _lacks_required_association(self, entry: IOCDeltaEntry) -> bool:
+        """Check whether an IOC entry lacks the required malware/threat actor associations.
+
+        The filter is disabled when both indicator_require_malware_family and
+        indicator_require_threat_actor are False (default). When one or both are
+        enabled, the indicator must have at least one matching association (OR
+        logic between the two). This filter is combined with the score filter
+        using AND logic.
+        """
+        require_malware = self.config.indicator_require_malware_family
+        require_threat_actor = self.config.indicator_require_threat_actor
+
+        if not require_malware and not require_threat_actor:
+            return False
+
+        rels = entry.relationships
+
+        has_malware = (
+            rels is not None
+            and rels.malware_families is not None
+            and len(rels.malware_families.data) > 0
+        )
+        has_threat_actor = (
+            rels is not None
+            and rels.threat_actors is not None
+            and len(rels.threat_actors.data) > 0
+        )
+
+        if require_malware and not require_threat_actor:
+            return not has_malware
+        if require_threat_actor and not require_malware:
+            return not has_threat_actor
+        # Both required: OR logic — at least one must be present
+        return not (has_malware or has_threat_actor)
+
     def convert(self, ioc_data: dict[str, Any]) -> list[Any]:
         """Convert a single IOC delta entry to STIX objects."""
         try:
@@ -97,6 +132,17 @@ def convert(self, ioc_data: dict[str, Any]) -> list[Any]:
             )
             return []
 
+        if self._lacks_required_association(entry):
+            self.logger.debug(
+                "IOC entry lacks required malware/threat actor association, skipping",
+                {
+                    "prefix": LOG_PREFIX,
+                    "type": entry.type,
+                    "id": entry.id,
+                },
+            )
+            return []
+
         try:
             self.logger.debug(
                 "Converting IOC delta entry to STIX",

external-import/google-ti-feeds/tests/custom/convert_to_stix/indicator/test_convert_to_stix_indicator.py

@@ -42,6 +42,8 @@ def __init__(self) -> None:  # noqa: D107
         self.indicator_types = ["file", "ip", "url", "domain"]
         self.indicator_import_start_date = timedelta(days=1)
         self.indicator_min_score = 0
+        self.indicator_require_malware_family = False
+        self.indicator_require_threat_actor = False
         self.report_import_start_date = timedelta(days=1)
         self.threat_actor_import_start_date = timedelta(days=1)
         self.malware_family_import_start_date = timedelta(days=1)
@@ -1048,3 +1050,243 @@ def test_min_score_none_disables_filter(self, logger: logging.Logger) -> None:
         result = _when_convert(converter, data)
         # Then – even a very low score entry is kept since the filter is disabled
         assert result  # noqa: S101
+
+
+# =====================
+# 27. convert – indicator_require_malware_family / indicator_require_threat_actor
+# =====================
+
+
+def _given_converter_with_association_filters(
+    logger: logging.Logger,
+    require_malware: bool = False,
+    require_threat_actor: bool = False,
+) -> ConvertToSTIXIndicator:
+    config = DummyConfig()
+    config.indicator_require_malware_family = require_malware
+    config.indicator_require_threat_actor = require_threat_actor
+    return ConvertToSTIXIndicator(
+        config=config,  # type: ignore[arg-type]
+        logger=logger,
+        tlp_level="white",
+    )
+
+
+def _given_relationships_with_malware_only() -> IOCDeltaRelationships:
+    return IOCDeltaRelationships(
+        malware_families=IOCDeltaRelationshipData(
+            data=[
+                IOCDeltaRelationshipItem(
+                    type="collection",
+                    id="malware--test",
+                    attributes=IOCDeltaRelationshipItemAttributes(name="TestMalware"),
+                )
+            ]
+        )
+    )
+
+
+def _given_relationships_with_threat_actor_only() -> IOCDeltaRelationships:
+    return IOCDeltaRelationships(
+        threat_actors=IOCDeltaRelationshipData(
+            data=[
+                IOCDeltaRelationshipItem(
+                    type="collection",
+                    id="threat-actor--test",
+                    attributes=IOCDeltaRelationshipItemAttributes(
+                        name="TestThreatActor"
+                    ),
+                )
+            ]
+        )
+    )
+
+
+def _given_relationships_with_both() -> IOCDeltaRelationships:
+    return IOCDeltaRelationships(
+        malware_families=IOCDeltaRelationshipData(
+            data=[
+                IOCDeltaRelationshipItem(
+                    type="collection",
+                    id="malware--test",
+                    attributes=IOCDeltaRelationshipItemAttributes(name="TestMalware"),
+                )
+            ]
+        ),
+        threat_actors=IOCDeltaRelationshipData(
+            data=[
+                IOCDeltaRelationshipItem(
+                    type="collection",
+                    id="threat-actor--test",
+                    attributes=IOCDeltaRelationshipItemAttributes(
+                        name="TestThreatActor"
+                    ),
+                )
+            ]
+        ),
+    )
+
+
+def _given_relationships_with_neither() -> IOCDeltaRelationships:
+    return IOCDeltaRelationships(
+        software_toolkits=IOCDeltaRelationshipData(
+            data=[
+                IOCDeltaRelationshipItem(
+                    type="collection",
+                    id="tool--test",
+                    attributes=IOCDeltaRelationshipItemAttributes(name="TestTool"),
+                )
+            ]
+        )
+    )
+
+
+class TestConvertAssociationFilter:
+    """Test convert() filtering IOC entries by association requirements."""
+
+    def test_no_filters_enabled_keeps_all(self, logger: logging.Logger) -> None:
+        # Given – both filters disabled (default)
+        converter = _given_converter_with_association_filters(logger)
+        data = _given_file_entry(relationships=_given_relationships_with_neither())
+        # When
+        result = _when_convert(converter, data)
+        # Then – entry is kept regardless of associations
+        assert result  # noqa: S101
+
+    def test_require_malware_with_malware_keeps(self, logger: logging.Logger) -> None:
+        # Given – require malware enabled, entry has malware association
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True
+        )
+        data = _given_file_entry(relationships=_given_relationships_with_malware_only())
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        assert result  # noqa: S101
+
+    def test_require_malware_without_malware_filters(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – require malware enabled, entry has no malware association
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True
+        )
+        data = _given_file_entry(
+            relationships=_given_relationships_with_threat_actor_only()
+        )
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        _then_returns_empty(result)
+
+    def test_require_threat_actor_with_threat_actor_keeps(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – require threat actor enabled, entry has threat actor association
+        converter = _given_converter_with_association_filters(
+            logger, require_threat_actor=True
+        )
+        data = _given_file_entry(
+            relationships=_given_relationships_with_threat_actor_only()
+        )
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        assert result  # noqa: S101
+
+    def test_require_threat_actor_without_threat_actor_filters(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – require threat actor enabled, entry has no threat actor association
+        converter = _given_converter_with_association_filters(
+            logger, require_threat_actor=True
+        )
+        data = _given_file_entry(relationships=_given_relationships_with_malware_only())
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        _then_returns_empty(result)
+
+    def test_both_required_with_malware_only_keeps(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – both filters enabled (OR logic), entry has malware only
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True, require_threat_actor=True
+        )
+        data = _given_file_entry(relationships=_given_relationships_with_malware_only())
+        # When
+        result = _when_convert(converter, data)
+        # Then – OR logic: malware is enough
+        assert result  # noqa: S101
+
+    def test_both_required_with_threat_actor_only_keeps(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – both filters enabled (OR logic), entry has threat actor only
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True, require_threat_actor=True
+        )
+        data = _given_file_entry(
+            relationships=_given_relationships_with_threat_actor_only()
+        )
+        # When
+        result = _when_convert(converter, data)
+        # Then – OR logic: threat actor is enough
+        assert result  # noqa: S101
+
+    def test_both_required_with_both_keeps(self, logger: logging.Logger) -> None:
+        # Given – both filters enabled, entry has both associations
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True, require_threat_actor=True
+        )
+        data = _given_file_entry(relationships=_given_relationships_with_both())
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        assert result  # noqa: S101
+
+    def test_both_required_with_neither_filters(self, logger: logging.Logger) -> None:
+        # Given – both filters enabled, entry has neither association
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True, require_threat_actor=True
+        )
+        data = _given_file_entry(relationships=_given_relationships_with_neither())
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        _then_returns_empty(result)
+
+    def test_require_malware_with_no_relationships_filters(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – require malware enabled, entry has no relationships at all
+        converter = _given_converter_with_association_filters(
+            logger, require_malware=True
+        )
+        data = _given_file_entry(relationships=None)
+        # When
+        result = _when_convert(converter, data)
+        # Then
+        _then_returns_empty(result)
+
+    def test_association_filter_combined_with_min_score(
+        self, logger: logging.Logger
+    ) -> None:
+        # Given – both score and association filters active (AND logic)
+        config = DummyConfig()
+        config.indicator_min_score = 50
+        config.indicator_require_malware_family = True
+        converter = ConvertToSTIXIndicator(
+            config=config,  # type: ignore[arg-type]
+            logger=logger,
+            tlp_level="white",
+        )
+        # Entry has malware association but score is below threshold
+        data = _given_file_entry(
+            score=10, relationships=_given_relationships_with_malware_only()
+        )
+        # When
+        result = _when_convert(converter, data)
+        # Then – AND logic: score filter rejects even with valid association
+        _then_returns_empty(result)

external-import/google-ti-feeds/tests/custom/orchestrators/indicator/test_orchestrator_indicator.py

@@ -89,6 +89,8 @@ def __init__(
         self.indicator_types = indicator_types
         self.indicator_import_start_date = indicator_import_start_date
         self.indicator_min_score = indicator_min_score
+        self.indicator_require_malware_family = False
+        self.indicator_require_threat_actor = False
         self.report_import_start_date = timedelta(days=1)
         self.threat_actor_import_start_date = timedelta(days=1)
         self.malware_family_import_start_date = timedelta(days=1)