diff --git a/stream/pan-cortex-xdr-intel/data_samples/octi_stream_message_example.json b/stream/pan-cortex-xdr-intel/data_samples/octi_stream_message_example.json new file mode 100644 index 00000000000..19aae4d7055 --- /dev/null +++ b/stream/pan-cortex-xdr-intel/data_samples/octi_stream_message_example.json @@ -0,0 +1,60 @@ +{ + "event": "create", + "id": "609198385", + "retry": null, + "data": { + "data": { + "id": "indicator--32c6886b-68e6-45e5-bb0f-0f5defa8b7ef", + "spec_version": "2.1", + "type": "indicator", + "extensions": { + "extension-definition--ea279b3e-5c71-4632-ac08-831c66a786ba": { + "extension_type": "property-extension", + "id": "15926f6f-44b9-446d-94e0-cf66c6aceec7", + "type": "Indicator", + "created_at": "2027-03-11T14:22:05.000Z", + "updated_at": "2027-03-11T14:39:05.000Z", + "is_inferred": false, + "creator_ids": [ + "7d9118e3-5f2f-453d-a20a-b4acd9a50e7a" + ], + "detection": false, + "score": 50, + "main_observable_type": "StixFile", + "observable_values": [ + { + "type": "StixFile", + "hashes": { + "MD5": "c2f7c2f12563246ea8ce5a4ea425de8e" + } + }, + { + "type": "StixFile", + "hashes": { + "SHA-256": "b59444a310faa5ab8a774161cd8961012e3c02daa70273bc75de8acfd8f504b7" + } + } + ] + }, + "extension-definition--322b8f77-262a-4cb8-a915-1e441e00329b": { + "extension_type": "property-extension" + } + }, + "created": "2027-03-11T14:22:05.000Z", + "modified": "2027-03-11T14:39:05.000Z", + "revoked": false, + "confidence": 100, + "lang": "en", + "name": "sample-indicator", + "pattern": "[file:hashes.MD5 = 'c2f7c2f12563246ea8ce5a4ea425de8e' AND file:hashes.'SHA-256' = 'b59444a310faa5ab8a774161cd8961012e3c02daa70273bc75de8acfd8f504b7']", + "pattern_type": "stix", + "valid_from": "2027-03-11T14:20:05.000Z", + "valid_until": "2027-12-26T14:22:05.000Z" + }, + "message": "creates a Indicator `sample-indicator`", + "origin": { + "referer": "init-create" + }, + "version": "4" + } +} diff --git a/stream/pan-cortex-xdr-intel/src/connector/connector.py b/stream/pan-cortex-xdr-intel/src/connector/connector.py index f0a6ded47d6..907b701deee 100644 --- a/stream/pan-cortex-xdr-intel/src/connector/connector.py +++ b/stream/pan-cortex-xdr-intel/src/connector/connector.py @@ -1,6 +1,10 @@ from __future__ import annotations -from typing import TYPE_CHECKING +import json +from typing import TYPE_CHECKING, Any, Protocol + +from connector.models import EventIndicator +from pydantic import ValidationError if TYPE_CHECKING: from connector.settings import ConnectorSettings @@ -8,6 +12,38 @@ from pycti import OpenCTIConnectorHelper +_SUPPORTED_EVENTS = {"create", "update", "delete"} +_SUPPORTED_ENTITY_TYPES = {"indicator"} +_SUPPORTED_OBSERVABLE_TYPES = { + "domain-name", + "hostname", + "ipv4-addr", + "ipv6-addr", + "stixfile", +} + +_OPENCTI_OBSERVABLE_TYPES_TO_XDR_IOC_TYPES = { + "domain-name": "DOMAIN_NAME", + "hostname": "DOMAIN_NAME", + "ipv4-addr": "IP", + "ipv6-addr": "IP", + "stixfile": {"name": "FILENAME", "hash": "HASH"}, + # XDR IOC types `PATH` and `MIXED` are not mapped for now +} + + +class StreamMessage(Protocol): + """Type for the SSE message passed to `listen_stream` callbacks. + + Only the attributes actually consumed by this connector are declared, + decoupling us from `filigran_sseclient.sseclient.Event`'s concrete shape + (and from adding it as an explicit dependency just for typing). + """ + + event: str + data: str + + class Connector: """ Cortex XDR Intel stream connector. @@ -38,11 +74,124 @@ def __init__( self.settings = settings self.client = client - def _process_message(self, message: dict) -> None: - # TODO: implement IOC upsert/delete mapping from the stream message (#7184/#7185) - self.helper.connector_logger.info( - "Received stream event (baseline wiring phase)" + # Reusable exit message for fatal errors logging + self._exit_message = ( + "Connector will exit to avoid further errors and/or exhausting the stream.\n" + "Please check connector's logs and report/fix the issue before restarting." + ) + + def _parse_indicator(self, data: dict[str, Any]) -> EventIndicator: + """Build a minimal, internal representation of an Indicator's stream + `data` payload. Field casting/validation is delegated to `EventIndicator`. + """ + # Use `observable_values` extension provided by OpenCTI to extract the list of observables + observable_values = ( + self.helper.get_attribute_in_extension("observable_values", data) or [] ) + # Filter out observable types not supported by Cortex XDR Client + supported_observable_values = [ + observable_value + for observable_value in observable_values + if observable_value.get("type", "").lower() in _SUPPORTED_OBSERVABLE_TYPES + ] + + # Parse and validate the indicator in stream `data` payload + return EventIndicator( + id=self.helper.get_attribute_in_extension("id", data), + description=data.get("description"), + observables=supported_observable_values, + valid_until=data.get("valid_until"), + score=self.helper.get_attribute_in_extension("score", data), + ) + + def _process_message(self, msg: StreamMessage) -> None: + """Process a single stream event message. + + Unsupported event or entity types are logged as a warning and + skipped. Any failure while decoding, parsing, or validating the + message (JSON decode error, `Indicator` validation error, or any + other unexpected exception) is logged with context and re-raised, + deliberately letting `pycti` kill the connector process rather than + risk silently missing or corrupting further events. + """ + event = msg.event + if event not in _SUPPORTED_EVENTS: + self.helper.connector_logger.warning( + "Unsupported event type, skipping it", + {"event": event}, + ) + return + + try: + message_data = json.loads(msg.data) + except json.JSONDecodeError as err: + # This should never happen and if it does, it indicates a breaking change in `pycti`. + # To avoid data loss, the connector must stop and the issue must be investigated and fixed before resuming. + self.helper.connector_logger.error( + f"Failed to parse stream event's `data` payload as JSON.\n{self._exit_message}", + { + "event": event, + "data": msg.data, + "error": err, + }, + ) + raise # let `pycti` kill the connector process + + entity_data = message_data.get("data", {}) + entity_type = entity_data.get("type") + if entity_type not in _SUPPORTED_ENTITY_TYPES: + self.helper.connector_logger.warning( + "Unsupported entity type, skipping it", + { + "event": event, + "entity_type": entity_type, + }, + ) + return + + try: + indicator = self._parse_indicator(entity_data) + + self.helper.connector_logger.info( + "Parsed observable(s) from stream event", + { + "event": event, + "indicator_id": indicator.id, + "observables_count": len(indicator.observables), + }, + ) + except ValidationError as err: + # This should never happen and if it does, it indicates a breaking change in `pycti`. + # To avoid data loss, the connector must stop and the issue must be investigated and fixed before resuming. + self.helper.connector_logger.error( + f"Failed to parse indicator and/or observables from stream event.\n{self._exit_message}", + { + "event": event, + "entity_data": entity_data, + "error": err, + }, + ) + raise # let `pycti` kill the connector process + + try: + if event in {"create", "update"}: + pass # TODO: upsert in Cortex XDR + elif event == "delete": + pass # TODO: delete from Cortex XDR + + except Exception as err: + # Repetitive unexpected errors could consume the stream in vain (no action performed). + # To avoid data loss, the connector must stop and the issue must be investigated and fixed before resuming. + self.helper.connector_logger.error( + f"Unexpected error while processing stream event.\n{self._exit_message}", + { + "event": event, + "entity_data": entity_data, + "error": err, + }, + ) + raise # let `pycti` kill the connector process def start(self) -> None: + """Start the connector's main loop: listen to the OpenCTI stream and process each message.""" self.helper.listen_stream(self._process_message) diff --git a/stream/pan-cortex-xdr-intel/src/connector/models.py b/stream/pan-cortex-xdr-intel/src/connector/models.py new file mode 100644 index 00000000000..16fb0f23f43 --- /dev/null +++ b/stream/pan-cortex-xdr-intel/src/connector/models.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class IndicatorObservable(BaseModel): + """A single observable extracted from an Indicator's `observable_values`.""" + + model_config = ConfigDict(frozen=True) + + type: str + value: str + + +class EventIndicator(BaseModel): + """Minimal, internal representation of an Indicator's stream `data` payload. + + This is a plain data carrier meant to decouple downstream handlers + (upsert/delete) from the raw OpenCTI STIX payload shape. Only `data` is + represented here: the stream action (create/update/delete) is a separate, + stream-level concern and is passed alongside this event, not stored on it. + Fields are cast/validated by pydantic to fail fast and minimize downstream + development/runtime errors. + """ + + model_config = ConfigDict(frozen=True) + + id: str + description: str | None = Field(default=None) + observables: list[IndicatorObservable] = Field(default_factory=list) + valid_until: datetime | None = Field(default=None) + score: int | None = Field(default=None) + + @field_validator("observables", mode="before") + def _validate_observables( + cls, value: list[dict[str, Any]] | None + ) -> list[dict[str, Any]]: + """Extract `observables` from the raw `observable_values` list returned by + `pycti.OpenCTIConnectorHelper.get_attribute_in_extension("observable_values", indicator)`. + + For `StixFile` observables, the filename (`name`) and each hash + algorithm value are extracted as separate observables, since Cortex + XDR treats them as distinct IOC types (`FILENAME` vs `HASH`). + """ + if not value: + return [] + + observables = [] + for observable_data in value: + observable_type = str(observable_data.get("type", "")) + if observable_type.lower() == "stixfile": + if name := observable_data.get("name"): + observables.append({"type": observable_type, "value": str(name)}) + for value in (observable_data.get("hashes") or {}).values(): + observables.append({"type": observable_type, "value": str(value)}) + elif value := observable_data.get("value"): + observables.append({"type": observable_type, "value": str(value)}) + + return observables diff --git a/stream/pan-cortex-xdr-intel/tests/tests_connector/test_connector.py b/stream/pan-cortex-xdr-intel/tests/tests_connector/test_connector.py new file mode 100644 index 00000000000..c8b7d6f7fdd --- /dev/null +++ b/stream/pan-cortex-xdr-intel/tests/tests_connector/test_connector.py @@ -0,0 +1,157 @@ +import json +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from connector.connector import Connector +from connector.models import IndicatorObservable +from pydantic import ValidationError + + +def _make_msg(event: str, data: dict) -> SimpleNamespace: + """Build a fake SSE stream event (attributes: `data`, `event`, `id`).""" + return SimpleNamespace(event=event, data=json.dumps({"data": data}), id="1-0") + + +@pytest.fixture +def connector(): + return Connector(helper=MagicMock(), settings=MagicMock(), client=MagicMock()) + + +class TestProcessMessageEventGuardrail: + def test_connector_skips_unsupported_event_without_parsing( + self, connector, monkeypatch + ): + # Given: a stream event whose action is not one of create/update/delete + parse_indicator = MagicMock() + monkeypatch.setattr(connector, "_parse_indicator", parse_indicator) + msg = _make_msg("invalid-event", {"type": "identity", "name": "Test Identity"}) + # When: processing the message + connector._process_message(msg) + # Then: no indicator parsing is attempted (the invalid JSON body is + # never even parsed) + parse_indicator.assert_not_called() + + +class TestProcessMessageEntityGuardrail: + def test_connector_skips_non_indicator_entity_without_parsing( + self, connector, monkeypatch + ): + # Given: a stream event for a non-Indicator entity (e.g. an Identity) + parse_indicator = MagicMock() + monkeypatch.setattr(connector, "_parse_indicator", parse_indicator) + msg = _make_msg("create", {"type": "identity", "name": "Test Identity"}) + # When: processing the message + connector._process_message(msg) + # Then: no indicator parsing is attempted + parse_indicator.assert_not_called() + + def test_connector_skips_event_missing_data_without_parsing( + self, connector, monkeypatch + ): + # Given: a stream event payload without a top-level "data" key + parse_indicator = MagicMock() + monkeypatch.setattr(connector, "_parse_indicator", parse_indicator) + msg = SimpleNamespace(event="create", data=json.dumps({}), id="1-0") + # When: processing the message + connector._process_message(msg) + # Then: it is treated as a non-Indicator event, no indicator parsing + # is attempted + parse_indicator.assert_not_called() + + +class TestParseIndicator: + def test_connector_extracts_supported_observable(self, connector): + # Given: an Indicator whose `observable_values` extension attribute + # contains a single MVP-supported (domain) observable + connector.helper.get_attribute_in_extension.side_effect = lambda key, data: { + "id": "indicator--id", + "observable_values": [{"type": "Domain-Name", "value": "evil.com"}], + }.get(key) + # When: parsing the indicator + indicator = connector._parse_indicator({"type": "indicator"}) + # Then: the observable is extracted + assert indicator.observables == [ + IndicatorObservable(type="Domain-Name", value="evil.com") + ] + + def test_connector_filters_out_unsupported_observable_type(self, connector): + # Given: an Indicator whose `observable_values` only contains an + # unsupported observable type (e.g. Mutex) + connector.helper.get_attribute_in_extension.side_effect = lambda key, data: { + "id": "indicator--id", + "observable_values": [{"type": "Mutex", "value": "some-mutex"}], + }.get(key) + # When: parsing the indicator + indicator = connector._parse_indicator({"type": "indicator"}) + # Then: no observable is extracted + assert indicator.observables == [] + + def test_connector_defaults_to_no_observables_when_extension_attribute_missing( + self, connector + ): + # Given: an Indicator without an `observable_values` extension attribute + # (e.g. older OpenCTI version, or non-STIX pattern indicator) + connector.helper.get_attribute_in_extension.side_effect = lambda key, data: { + "id": "indicator--id", + "observable_values": None, + }.get(key) + # When: parsing the indicator + indicator = connector._parse_indicator({"type": "indicator"}) + # Then: no observable is extracted, no crash + assert indicator.observables == [] + + +class TestProcessMessageErrorHandling: + # Fatal errors + + def test_connector_logs_error_and_reraises_on_invalid_json(self, connector): + # Given: a stream event whose `data` is not valid JSON + msg = SimpleNamespace(event="create", data="not-json", id="1-0") + # When: processing the message + # Then: the error is logged with context, and the exception is deliberately + # re-raised to let `pycti` kill the connector process + with pytest.raises(json.JSONDecodeError): + connector._process_message(msg) + connector.helper.connector_logger.error.assert_called_once() + + def test_connector_logs_error_and_reraises_when_indicator_id_missing( + self, connector + ): + # Given: an Indicator stream event missing its "id" extension attribute + # (required by `EventIndicator`), causing `_parse_indicator` to raise + # a `ValidationError` + connector.helper.get_attribute_in_extension.side_effect = lambda key, data: { + "id": None, + "observable_values": None, + }.get(key) + msg = _make_msg("create", {"type": "indicator"}) + # When: processing the message + # Then: the error is logged with context, and the exception is deliberately + # re-raised to let `pycti` kill the connector process + with pytest.raises(ValidationError): + connector._process_message(msg) + connector.helper.connector_logger.error.assert_called_once() + + @pytest.mark.xfail( + reason=( + "Upsert/delete client calls are not implemented yet (see #7186/#7187); " + "the try/except around them currently only wraps `pass` placeholders, " + "so this expected error-handling behavior cannot be exercised yet. " + "Remove this xfail marker once real client calls land in that block." + ), + strict=True, + ) + def test_connector_logs_error_and_reraises_on_unexpected_processing_error( + self, connector + ): + # Given: the Cortex XDR client unexpectedly raises while upserting/deleting + connector.client.upsert_indicator.side_effect = RuntimeError("boom") + connector.client.delete_indicator.side_effect = RuntimeError("boom") + msg = _make_msg("create", {"type": "indicator"}) + # When: processing the message + # Then: the error is logged with context, and the exception is deliberately + # re-raised to let `pycti` kill the connector process + with pytest.raises(RuntimeError, match="boom"): + connector._process_message(msg) + connector.helper.connector_logger.error.assert_called_once() diff --git a/stream/pan-cortex-xdr-intel/tests/tests_connector/test_models.py b/stream/pan-cortex-xdr-intel/tests/tests_connector/test_models.py new file mode 100644 index 00000000000..9b62207da17 --- /dev/null +++ b/stream/pan-cortex-xdr-intel/tests/tests_connector/test_models.py @@ -0,0 +1,81 @@ +from datetime import datetime, timezone + +import pytest +from connector.models import EventIndicator, IndicatorObservable +from pydantic import ValidationError + + +class TestEventIndicator: + """Happy/unhappy path coverage for `EventIndicator` construction, including + its `observables` field_validator - the only custom logic in `models.py`, + which normalizes OpenCTI's raw `observable_values` extension attribute. + """ + + def test_event_indicator_model_accepts_valid_input(self): + # Given: a valid, complete set of fields for an EventIndicator + # When: constructing an EventIndicator + indicator = EventIndicator( + id="indicator--id", + description="Malicious domain", + observables=[{"type": "Domain-Name", "value": "evil.com"}], + valid_until="2030-01-01T00:00:00Z", + score=75, + ) + # Then: the indicator is constructed successfully, fields cast to + # their expected Python type + assert indicator.id == "indicator--id" + assert indicator.description == "Malicious domain" + assert indicator.valid_until == datetime(2030, 1, 1, 0, 0, tzinfo=timezone.utc) + assert indicator.score == 75 + assert indicator.observables == [ + IndicatorObservable(type="Domain-Name", value="evil.com") + ] + + def test_event_indicator_model_rejects_missing_id(self): + # Given: no `id` value (e.g. missing extension attribute) + # When: constructing an EventIndicator + # Then: a ValidationError is raised, reporting `id` as required + with pytest.raises(ValidationError, match=r"id\s+Field required"): + EventIndicator() + + def test_event_indicator_model_flattens_stixfile_hashes_into_one_observable_per_algorithm( + self, + ): + # Given: a raw StixFile observable with multiple hash algorithms + # When: constructing an EventIndicator + indicator = EventIndicator( + id="indicator--id", + observables=[ + {"type": "StixFile", "hashes": {"MD5": "aaa", "SHA-256": "bbb"}} + ], + ) + # Then: one observable per hash algorithm is extracted + assert indicator.observables == [ + IndicatorObservable(type="StixFile", value="aaa"), + IndicatorObservable(type="StixFile", value="bbb"), + ] + + def test_event_indicator_model_skips_stixfile_observable_without_hashes(self): + # Given: a raw StixFile observable without a "hashes" key + # When: constructing an EventIndicator + indicator = EventIndicator( + id="indicator--id", observables=[{"type": "StixFile"}] + ) + # Then: no observable is extracted + assert indicator.observables == [] + + def test_event_indicator_model_skips_observable_without_value(self): + # Given: a raw observable missing its "value" key + # When: constructing an EventIndicator + indicator = EventIndicator( + id="indicator--id", observables=[{"type": "Domain-Name"}] + ) + # Then: no observable is extracted + assert indicator.observables == [] + + def test_event_indicator_model_defaults_observables_to_empty_list_when_none(self): + # Given: observables explicitly set to None (e.g. missing extension attribute) + # When: constructing an EventIndicator + indicator = EventIndicator(id="indicator--id", observables=None) + # Then: observables default to an empty list, no error + assert indicator.observables == []