[connectors-sdk] Compute BaseIdentifiedObject.id on validation and cache its value (#6034)

Author: Powlinett

connectors-sdk/connectors_sdk/models/base_identified_entity.py

@@ -10,17 +10,12 @@
 from connectors_sdk.models.external_reference import ExternalReference
 from connectors_sdk.models.reference import Reference
 from connectors_sdk.models.tlp_marking import TLPMarking
-from pydantic import (
-    Field,
-    PrivateAttr,
-)
+from pydantic import Field
 
 
 class BaseIdentifiedEntity(BaseIdentifiedObject, ABC):
     """Base class that can be identified thanks to a stix-like id."""
 
-    _stix2_id: str | None = PrivateAttr(default=None)
-
     author: BaseAuthorEntity | Reference | None = Field(
         default=None,
         description="The Author reporting this Observable.",

connectors-sdk/connectors_sdk/models/base_identified_object.py

@@ -1,11 +1,9 @@
 """BaseIdentifiedObject."""
 
-from __future__ import annotations
-
 import warnings
 from abc import ABC
-from typing import Any
 
+import stix2
 from connectors_sdk.models.base_object import BaseObject
 from pydantic import (
     PrivateAttr,
@@ -19,37 +17,34 @@ class BaseIdentifiedObject(BaseObject, ABC):
 
     _stix2_id: str | None = PrivateAttr(default=None)
 
-    def model_post_init(self, context__: Any) -> None:
-        """Define the post initialization method, automatically called after __init__ in a pydantic model initialization.
+    @computed_field  # type: ignore[prop-decorator]
+    # Typing known issue : see https://docs.pydantic.dev/2.3/usage/computed_fields/ (consulted on 2025-06-06)
+    @property
+    def id(self) -> str:
+        """Return the unique identifier of the entity.
+        The value is computed from the STIX2 object representation of the entity, and then cached for future use.
 
         Notes:
-            This allows a last modification of the pydantic Model before it is validated.
-
-        Args:
-            context__(Any): The pydantic context used by pydantic framework.
-
-        References:
-            https://docs.pydantic.dev/latest/api/base_model/#pydantic.BaseModel.model_parametrized_name [consulted on
-                October 4th, 2024]
-
+            The decorator `@computed_field` is used to indicate that this property
+            must be considered as a field by pydantic, and then included in the model **serialization** processes.
+            This field is **not** validated by pydantic, neither during the model initialization, nor during the model update.
         """
-        _ = context__  # Unused parameter, but required by pydantic
         if self._stix2_id is None:
-            self._stix2_id = self.id
+            stix_object: stix2.v21._STIXBase21 = self.to_stix2_object()
+            stix_id = stix_object.get("id")
 
-    @computed_field  # type: ignore[prop-decorator]
-    # known issue : see https://docs.pydantic.dev/2.3/usage/computed_fields/ (consulted on 2025-06-06)
-    @property
-    def id(self) -> str:
-        """Return the unique identifier of the entity."""
-        stix_id: str = self.to_stix2_object().get("id", "")
-        self._stix2_id = stix_id
-        return stix_id
+            # The 'id' property is required and must be a non-empty string for model validation
+            if not (isinstance(stix_id, str) and stix_id.strip()):
+                raise ValueError("The 'id' property can't be set.")
+
+            self._stix2_id = stix_id
+
+        return self._stix2_id
 
     # https://github.com/pydantic/pydantic/discussions/10098
     @model_validator(mode="after")
-    def _check_id(self) -> BaseIdentifiedObject:
-        """Ensure the id is correctly set and alert if it has changed.
+    def _compute_stix_id(self) -> "BaseIdentifiedObject":
+        """Compute STIX ID on validation (instance creation or re-assignments).
 
         Raises:
             ValueError: If the id is not set.
@@ -73,19 +68,23 @@ def _check_id(self) -> BaseIdentifiedObject:
             'identity--011fe1ae-7b92-4779-9eb5-7be2aeffb9e1'
 
         """
-        if self._stix2_id is None or self._stix2_id == "":
-            raise ValueError("The 'id' property must be set.")
-
-        if self._stix2_id != self.id:
-            # define message before the warning to avoid self._stix2_id has already changed in the main thread
-            message = (
-                f"The 'id' property has changed from to {self.id}. "
-                "This may lead to unexpected behavior in the OpenCTI platform."
-            )
+        previous_stix2_id = self._stix2_id
+        self._stix2_id = None  # Reset cached id
+        current_stix2_id = self.id  # Compute and cache the new id
+
+        if previous_stix2_id is None:
+            # If the previous id was not set, just set it without warning
+            return self
+
+        if previous_stix2_id != current_stix2_id:
+            # Warn on `_stix2_id` change after fields re-assignment
             warnings.warn(
-                message=message,
+                message=(
+                    f"The 'id' property has changed from {previous_stix2_id} to {current_stix2_id}. "
+                    "This may lead to unexpected behavior in the connector or the OpenCTI platform."
+                ),
                 category=UserWarning,
                 stacklevel=2,
             )
-        self._stix2_id = self.id  # Update the internal id to the current one
+
         return self

internal-enrichment/censys-enrichment/src/censys_enrichment/converter.py

@@ -12,7 +12,6 @@
 from connectors_sdk.models import (
     AdministrativeArea,
     AutonomousSystem,
-    BaseIdentifiedEntity,
     BaseObject,
     City,
     Country,
@@ -22,35 +21,14 @@
     Note,
     Organization,
     OrganizationAuthor,
+    Reference,
     Region,
     Relationship,
     Software,
     TLPMarking,
     X509Certificate,
 )
 from connectors_sdk.models.enums import HashAlgorithm, RelationshipType, TLPLevel
-from pydantic import Field
-
-
-class EmbeddedIdentifiedStixObject(BaseIdentifiedEntity):
-    """Embedded Identified STIX Object representation.
-
-    This class encapsulates a STIX object with an id as a dictionary and provides
-    access to the object without copying or modifying the original data.
-
-    Use when you only need to read or forward the STIX object, not alter it.
-    """
-
-    stix_object: dict[str, Any] = Field()
-
-    @property
-    def id(self) -> str:
-        """Return the STIX object's ID."""
-        return self.stix_object["id"]
-
-    def to_stix2_object(self) -> dict[str, Any]:
-        """Return the STIX2 object representation."""
-        return self.stix_object
 
 
 class Converter:
@@ -60,7 +38,7 @@ def __init__(self) -> None:
         self._common_props = {"author": self.author, "markings": [self.marking]}
 
     def _generate_city(
-        self, observable: EmbeddedIdentifiedStixObject, name: str | None
+        self, observable: Reference, name: str | None
     ) -> Generator[BaseObject, None, None]:
         if not name:
             return
@@ -80,7 +58,7 @@ def _generate_city(
         ]
 
     def _generate_country(
-        self, observable: EmbeddedIdentifiedStixObject, name: str | None
+        self, observable: Reference, name: str | None
     ) -> Generator[BaseObject, None, Country | None]:
         if not name:
             return None
@@ -101,7 +79,7 @@ def _generate_country(
         return country
 
     def _generate_region(
-        self, observable: EmbeddedIdentifiedStixObject, name: str | None
+        self, observable: Reference, name: str | None
     ) -> Generator[BaseObject, None, None]:
         if not name:
             return
@@ -122,7 +100,7 @@ def _generate_region(
 
     def _generate_administrative_area(
         self,
-        observable: EmbeddedIdentifiedStixObject,
+        observable: Reference,
         name: str | None,
         coordinates: Coordinates | None,
     ) -> Generator[BaseObject, None, None]:
@@ -154,7 +132,7 @@ def _generate_administrative_area(
         ]
 
     def _generate_hostnames(
-        self, observable: EmbeddedIdentifiedStixObject, dns: HostDNS | None
+        self, observable: Reference, dns: HostDNS | None
     ) -> Generator[BaseObject, None, None]:
         if not dns:
             return
@@ -176,7 +154,7 @@ def _generate_hostnames(
 
     def _generate_organization(
         self,
-        observable: EmbeddedIdentifiedStixObject,
+        observable: Reference,
         name: str | None,
     ) -> Generator[BaseObject, None, Organization | None]:
         if not name:
@@ -199,7 +177,7 @@ def _generate_organization(
 
     def _generate_autonomous_system(
         self,
-        observable: EmbeddedIdentifiedStixObject,
+        observable: Reference,
         number: int | None,
         name: str | None,
         description: str | None,
@@ -226,7 +204,7 @@ def _generate_autonomous_system(
 
     def _generate_software(
         self,
-        observable: EmbeddedIdentifiedStixObject,
+        observable: Reference,
         name: str | None,
         vendor: str | None,
         cpe: str | None,
@@ -317,7 +295,7 @@ def _generate_certificate(
 
     def _generate_note(
         self,
-        observable: EmbeddedIdentifiedStixObject,
+        observable: Reference,
         content: str | None,
         publication_date: str | None,
         port: int | None,
@@ -335,7 +313,7 @@ def _generate_note(
         )
 
     def _generate_services(
-        self, observable: EmbeddedIdentifiedStixObject, services: list[Service] | None
+        self, observable: Reference, services: list[Service] | None
     ) -> Generator[BaseObject, None, None]:
         for service in services or []:
             for software in service.software or []:
@@ -363,7 +341,7 @@ def _generate_services(
             )
 
     def _generate_ip(
-        self, observable: EmbeddedIdentifiedStixObject, ip: str
+        self, observable: Reference, ip: str
     ) -> Generator[BaseObject, None, None | IPV4Address | IPV6Address]:
         ip_version = ipaddress.ip_network(ip, strict=False).version
         if ip_version == 4:
@@ -384,7 +362,7 @@ def _generate_ip(
     def generate_octi_objects(
         self, stix_entity: dict[str, Any], data: Host
     ) -> Generator[BaseObject, None, None]:
-        observable = EmbeddedIdentifiedStixObject(stix_object=stix_entity)
+        observable = Reference(id=stix_entity.get("id"))
 
         yield from [
             self.author,
@@ -461,7 +439,7 @@ def generate_octi_objects_from_hosts(
     ) -> Generator[BaseObject, None, None]:
         for host in hosts:
             ip_stix = yield from self._generate_ip(
-                observable=EmbeddedIdentifiedStixObject(stix_object=stix_entity),
+                observable=Reference(id=stix_entity.get("id")),
                 ip=host.ip,
             )
             yield from self.generate_octi_objects(