diff --git a/connectors-sdk/connectors_sdk/settings/_settings_loader.py b/connectors-sdk/connectors_sdk/settings/_settings_loader.py new file mode 100644 index 00000000000..d7b3197cd6f --- /dev/null +++ b/connectors-sdk/connectors_sdk/settings/_settings_loader.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import sys +from copy import deepcopy +from pathlib import Path +from types import UnionType +from typing import TYPE_CHECKING, Any, Union, get_args, get_origin + +from pydantic import BaseModel, create_model +from pydantic_settings import ( + BaseSettings, + DotEnvSettingsSource, + PydanticBaseSettingsSource, + SettingsConfigDict, + YamlConfigSettingsSource, +) + +if TYPE_CHECKING: + from connectors_sdk.settings.base_settings import BaseConnectorSettings + + +class _SettingsLoader(BaseSettings): + model_config = SettingsConfigDict( + frozen=True, + extra="allow", + env_nested_delimiter="_", + env_nested_max_split=1, + enable_decoding=False, + ) + + @classmethod + def _get_connector_main_path(cls) -> Path: + """Locate the main module of the running connector. + This method is used to locate configuration files relative to connector's entrypoint. + + Notes: + - This method assumes that the connector is launched using a file-backed entrypoint + (i.e., `python -m ` or `python `). + - At module import time, `__main__.__file__` might not be available yet, + thus this method should be called at runtime only. + """ + main = sys.modules.get("__main__") + if main and getattr(main, "__file__", None): + return Path(main.__file__).resolve() # type: ignore + + raise RuntimeError( + "Cannot determine connector's location: __main__.__file__ is not available. " + "Ensure the connector is launched using `python -m ` or a file-backed entrypoint." + ) + + @classmethod + def _get_config_yml_file_path(cls) -> Path | None: + """Locate the `config.yml` file of the running connector.""" + main_path = cls._get_connector_main_path() + config_yml_legacy_file_path = main_path.parent / "config.yml" + config_yml_file_path = main_path.parent.parent / "config.yml" + + if config_yml_legacy_file_path.is_file(): + return config_yml_legacy_file_path + elif config_yml_file_path.is_file(): + return config_yml_file_path + return None + + @classmethod + def _get_dot_env_file_path(cls) -> Path | None: + """Locate the `.env` file of the running connector.""" + main_path = cls._get_connector_main_path() + dot_env_file_path = main_path.parent.parent / ".env" + + return dot_env_file_path if dot_env_file_path.is_file() else None + + @classmethod + def settings_customise_sources( + cls, + settings_cls: type[BaseSettings], + init_settings: PydanticBaseSettingsSource, + env_settings: PydanticBaseSettingsSource, + dotenv_settings: PydanticBaseSettingsSource, + file_secret_settings: PydanticBaseSettingsSource, + ) -> tuple[PydanticBaseSettingsSource, ...]: + """Customise the sources of settings for the connector. + + This method is called by the Pydantic BaseSettings class to determine the order of sources. + The configuration come in this order either from: + 1. Environment variables + 2. YAML file + 3. .env file + 4. Default values + + The variables loading order will remain the same as in `pycti.get_config_variable()`: + 1. If a config.yml file is found, the order will be: `ENV VAR` → config.yml → default value + 2. If a .env file is found, the order will be: `ENV VAR` → .env → default value + """ + config_yml_file_path = cls._get_config_yml_file_path() + if config_yml_file_path: + return ( + env_settings, + YamlConfigSettingsSource(settings_cls, yaml_file=config_yml_file_path), + ) + + dot_env_file_path = cls._get_dot_env_file_path() + if dot_env_file_path: + return ( + env_settings, + DotEnvSettingsSource(settings_cls, env_file=dot_env_file_path), + ) + + return (env_settings,) + + @classmethod + def build_loader_from_model( + cls, connector_settings: type[BaseConnectorSettings] + ) -> type[_SettingsLoader]: + """Build an untyped `_SettingsLoader` subclass for a connector's settings. + + This method dynamically creates a subclass of `_SettingsLoader` that mirrors the + structure of the provided `BaseConnectorSettings` implementation. It disables all + Pydantic decoding, type coercion and validation so fields accept raw, unprocessed values. + + The resulting model: + * Preserves values as-is from configuration sources + * Keeps YAML values as native Python types + * Keeps environment variables as plain strings + * Allows any field type (`Any`) without validation + + Args: + connector_settings (type[BaseConnectorSettings]): The typed connector settings class to mirror. + + Returns: + type[_SettingsLoader]: A dynamically generated subclass of `_SettingsLoader` + where all fields accept raw, unvalidated input. + """ + + class SettingsLoader(_SettingsLoader): ... + + model_fields = deepcopy(connector_settings.model_fields) + for field_info in model_fields.values(): + annotation = field_info.annotation + + # Unwrap `BaseModel | None` / `Optional[BaseModel]` annotations + annotation_origin = get_origin(annotation) + if annotation_origin in (Union, UnionType): + base_model_annotation = next( + ( + arg + for arg in get_args(annotation) + if isinstance(arg, type) and issubclass(arg, BaseModel) + ), + None, + ) + if base_model_annotation: + annotation = base_model_annotation + + # Keep only `BaseModel` model fields names (accept any value) + if isinstance(annotation, type) and issubclass(annotation, BaseModel): + fields: dict[str, Any] = dict.fromkeys( + annotation.model_fields.keys(), Any + ) + untyped_model = create_model( + f"{annotation.__name__}Untyped", + __base__=annotation, + **fields, + ) + field_info.annotation = untyped_model + field_info.default_factory = untyped_model + + SettingsLoader.model_fields = model_fields # type: ignore + return SettingsLoader diff --git a/connectors-sdk/connectors_sdk/settings/annotated_types.py b/connectors-sdk/connectors_sdk/settings/annotated_types.py index 880a53f9eaf..be0474ef338 100644 --- a/connectors-sdk/connectors_sdk/settings/annotated_types.py +++ b/connectors-sdk/connectors_sdk/settings/annotated_types.py @@ -6,7 +6,6 @@ from pydantic import ( BeforeValidator, PlainSerializer, - SerializationInfo, TypeAdapter, ) @@ -35,58 +34,17 @@ def parse_comma_separated_list(value: str | list[str]) -> list[str]: return value -def serialize_list_of_strings( - value: list[str], info: SerializationInfo -) -> str | list[str]: - """Serialize a list[str] as a comma-separated string when the Pydantic - serialization context requests "pycti" mode; otherwise, return the list - unchanged. - - This serializer is intended for use with Pydantic v2 `PlainSerializer` and - is typically activated only during JSON serialization (`when_used="json"`), - so the in-memory Python value remains a `list[str]` while the JSON output - can be a single string when required by external systems. - - Parameters - - value: The value to serialize. Expected to be a list of strings. - - info: Serialization context provided by Pydantic. If `info.context` - contains `{"mode": "pycti"}`, the list will be joined into a single - comma-separated string. - - Returns: - - A comma-separated string if context mode is "pycti" and `value` is a list. - - The original value `value` unchanged in all other cases. - - Notes: - - Joining does not insert spaces; e.g., ["a", "b", "c"] -> "a,b,c". - - If any element contains commas, those commas are not escaped. - - Examples: - - info.context={"mode": "pycti"} and value=["e1", "e2"] -> "e1,e2" - - info.context is None or mode != "pycti" -> ["e1", "e2"] - """ - if info.context and info.context.get("mode") == "pycti": - return ",".join(value) # [ "e1", "e2", "e3" ] -> "e1,e2,e3" - return value - - ListFromString = Annotated[ list[str], # Final type BeforeValidator(parse_comma_separated_list), - PlainSerializer(serialize_list_of_strings, when_used="json"), """Annotated list[str] that: - Validates: Accepts a comma-separated string (e.g., "a,b,c") or a list[str]. If a string is provided, it is split on commas and whitespace is trimmed for each item. -- Serializes (JSON): When the Pydantic serialization context includes - {"mode": "pycti"}, the list is serialized as a single comma-separated string - (e.g., ["a","b"] -> "a,b"). Otherwise, it serializes as a JSON array by default. Components - BeforeValidator(parse_comma_separated_list): Converts input strings to list[str] early in validation. -- PlainSerializer(serialize_list_of_strings, when_used="json"): Produces the "pycti" - string form only for JSON serialization. Examples - Validation: @@ -97,12 +55,6 @@ class Model(BaseModel): Model.model_validate({"tags": "a, b , c"}).tags # -> ["a", "b", "c"] Model.model_validate({"tags": ["x", "y"]}).tags # -> ["x", "y"] - -- Serialization: - m = Model.model_validate({"tags": ["e1", "e2"]}) - m.model_dump() # -> {'tags': ['e1', 'e2']} - m.model_dump_json() # -> {"tags":["e1","e2"]} - m.model_dump_json(context={"mode": "pycti"}) # -> {"tags":"e1,e2"} """, ] @@ -172,6 +124,5 @@ class Model(BaseModel): m = Model.model_validate({"start_date": datetime(2023, 10, 01, 0, 0, tzinfo=timezone.utc)}) m.model_dump() # -> {'start_date': datetime(2023, 10, 01, 0, 0, tzinfo=timezone.utc)} m.model_dump_json() # -> {"start_date": "2023-10-01T00:00:00+00:00"} - m.model_dump_json(context={"mode": "pycti"}) # -> {"start_date": "2023-10-01T00:00:00+00:00"} """, ] diff --git a/connectors-sdk/connectors_sdk/settings/base_settings.py b/connectors-sdk/connectors_sdk/settings/base_settings.py index 961c3c2e2af..a6d13ab895b 100644 --- a/connectors-sdk/connectors_sdk/settings/base_settings.py +++ b/connectors-sdk/connectors_sdk/settings/base_settings.py @@ -6,13 +6,12 @@ These models can be extended to create specific configurations for different types of connectors. """ -import sys from abc import ABC -from copy import deepcopy from datetime import timedelta -from pathlib import Path -from typing import Any, ClassVar, Literal, Self +from types import UnionType +from typing import Any, ClassVar, Literal, Self, Union, get_args, get_origin +from connectors_sdk.settings._settings_loader import _SettingsLoader from connectors_sdk.settings.annotated_types import ListFromString from connectors_sdk.settings.deprecations import ( Deprecate, @@ -27,20 +26,16 @@ BaseModel, ConfigDict, Field, + FieldSerializationInfo, HttpUrl, ModelWrapValidatorHandler, + SecretStr, + SerializerFunctionWrapHandler, ValidationError, - create_model, + field_serializer, model_validator, ) from pydantic.fields import FieldInfo -from pydantic_settings import ( - BaseSettings, - DotEnvSettingsSource, - PydanticBaseSettingsSource, - SettingsConfigDict, - YamlConfigSettingsSource, -) class BaseConfigModel(BaseModel, ABC): @@ -48,7 +43,11 @@ class BaseConfigModel(BaseModel, ABC): To prevent attributes from being modified after initialization. """ - model_config = ConfigDict(extra="allow", frozen=True, validate_default=True) + model_config = ConfigDict( + extra="allow", + frozen=True, + validate_default=True, + ) _model_deprecated_fields: ClassVar[dict[str, FieldInfo]] = {} @@ -62,13 +61,21 @@ def __pydantic_init_subclass__(cls, **kwargs: Any) -> None: for name, field in cls.model_fields.items(): for meta in field.metadata: if isinstance(meta, Deprecate): - # Change validation behavior - if not field.deprecated: - field.deprecated = True + # Make the field optional (accept `None`) + annotation = field.annotation + if isinstance(annotation, type): + field.annotation = annotation | None # type: ignore[assignment] + elif annotation is not None and get_origin(annotation) is not None: + # Handle typing wrappers (e.g. SkipJsonSchema[X], Annotated[X, ...]) + field.annotation = annotation | None field.default = None field.default_factory = None field.validate_default = False + # Mark as deprecated (in case of missing/empty deprecation message) + if not field.deprecated: + field.deprecated = True + # Add deprecation info to JSON schema if not field.json_schema_extra: field.json_schema_extra = {} @@ -90,10 +97,25 @@ class _OpenCTIConfig(BaseConfigModel): url: HttpUrl = Field( description="The base URL of the OpenCTI instance.", ) - token: str = Field( + token: SecretStr = Field( description="The API token to connect to OpenCTI.", ) + @field_serializer("token", mode="wrap", when_used="json") + def _serialize_token( + self, + value: Any, + handler: SerializerFunctionWrapHandler, + info: FieldSerializationInfo, + ) -> str: + """Get token secret value when serializing for `pycti.OpenCTIConnectorHelper` only. + Otherwise, return the redacted value, i.e. `"**********"`. + """ + mode = info.context.get("mode") if info.context else None + if isinstance(value, SecretStr) and mode == "pycti": + return value.get_secret_value() + return handler(value) # type: ignore[no-any-return] # actually return `str` + class _BaseConnectorConfig(BaseConfigModel, ABC): """Base class for connector configuration. @@ -113,145 +135,27 @@ class _BaseConnectorConfig(BaseConfigModel, ABC): description="The name of the connector.", ) scope: ListFromString = Field( - description="The scope of the connector, e.g. 'flashpoint'." + description="The scope of the connector, e.g. 'indicator, vulnerability'.", ) log_level: Literal["debug", "info", "warn", "warning", "error"] = Field( description="The minimum level of logs to display.", default="error", ) - -class _SettingsLoader(BaseSettings): - model_config = SettingsConfigDict( - frozen=True, - extra="allow", - env_nested_delimiter="_", - env_nested_max_split=1, - enable_decoding=False, - ) - - @classmethod - def _get_connector_main_path(cls) -> Path: - """Locate the main module of the running connector. - This method is used to locate configuration files relative to connector's entrypoint. - - Notes: - - This method assumes that the connector is launched using a file-backed entrypoint - (i.e., `python -m ` or `python `). - - At module import time, `__main__.__file__` might not be available yet, - thus this method should be called at runtime only. - """ - main = sys.modules.get("__main__") - if main and getattr(main, "__file__", None): - return Path(main.__file__).resolve() # type: ignore - - raise RuntimeError( - "Cannot determine connector's location: __main__.__file__ is not available. " - "Ensure the connector is launched using `python -m ` or a file-backed entrypoint." - ) - - @classmethod - def _get_config_yml_file_path(cls) -> Path | None: - """Locate the `config.yml` file of the running connector.""" - main_path = cls._get_connector_main_path() - config_yml_legacy_file_path = main_path.parent / "config.yml" - config_yml_file_path = main_path.parent.parent / "config.yml" - - if config_yml_legacy_file_path.is_file(): - return config_yml_legacy_file_path - elif config_yml_file_path.is_file(): - return config_yml_file_path - return None - - @classmethod - def _get_dot_env_file_path(cls) -> Path | None: - """Locate the `.env` file of the running connector.""" - main_path = cls._get_connector_main_path() - dot_env_file_path = main_path.parent.parent / ".env" - - return dot_env_file_path if dot_env_file_path.is_file() else None - - @classmethod - def settings_customise_sources( - cls, - settings_cls: type[BaseSettings], - init_settings: PydanticBaseSettingsSource, - env_settings: PydanticBaseSettingsSource, - dotenv_settings: PydanticBaseSettingsSource, - file_secret_settings: PydanticBaseSettingsSource, - ) -> tuple[PydanticBaseSettingsSource, ...]: - """Customise the sources of settings for the connector. - - This method is called by the Pydantic BaseSettings class to determine the order of sources. - The configuration come in this order either from: - 1. Environment variables - 2. YAML file - 3. .env file - 4. Default values - - The variables loading order will remain the same as in `pycti.get_config_variable()`: - 1. If a config.yml file is found, the order will be: `ENV VAR` → config.yml → default value - 2. If a .env file is found, the order will be: `ENV VAR` → .env → default value - """ - config_yml_file_path = cls._get_config_yml_file_path() - if config_yml_file_path: - return ( - env_settings, - YamlConfigSettingsSource(settings_cls, yaml_file=config_yml_file_path), - ) - - dot_env_file_path = cls._get_dot_env_file_path() - if dot_env_file_path: - return ( - env_settings, - DotEnvSettingsSource(settings_cls, env_file=dot_env_file_path), - ) - - return (env_settings,) - - @classmethod - def build_loader_from_model( - cls, connector_settings: type["BaseConnectorSettings"] - ) -> type["_SettingsLoader"]: - """Build an untyped `_SettingsLoader` subclass for a connector's settings. - - This method dynamically creates a subclass of `_SettingsLoader` that mirrors the - structure of the provided `BaseConnectorSettings` implementation. It disables all - Pydantic decoding, type coercion and validation so fields accept raw, unprocessed values. - - The resulting model: - * Preserves values as-is from configuration sources - * Keeps YAML values as native Python types - * Keeps environment variables as plain strings - * Allows any field type (`Any`) without validation - - Args: - connector_settings (type[BaseConnectorSettings]): The typed connector settings class to mirror. - - Returns: - type[_SettingsLoader]: A dynamically generated subclass of `_SettingsLoader` - where all fields accept raw, unvalidated input. + @field_serializer("scope", mode="wrap", when_used="json") + def _serialize_scope( + self, + value: Any, + handler: SerializerFunctionWrapHandler, + info: FieldSerializationInfo, + ) -> str | list[str]: + """Serialize scope as a comma-separated string when serializing for `pycti.OpenCTIConnectorHelper` only. + Otherwise, return the list of strings. """ - - class SettingsLoader(_SettingsLoader): ... - - model_fields = deepcopy(connector_settings.model_fields) - for field_info in model_fields.values(): - annotation = field_info.annotation - if annotation and issubclass(annotation, BaseModel): - fields: dict[str, Any] = dict.fromkeys( - annotation.model_fields.keys(), Any - ) - untyped_model = create_model( - f"{annotation.__name__}Untyped", - __base__=annotation, - **fields, - ) - field_info.annotation = untyped_model - field_info.default_factory = untyped_model - - SettingsLoader.model_fields = model_fields # type: ignore - return SettingsLoader + mode = info.context.get("mode") if info.context else None + if isinstance(value, list) and mode == "pycti": + return ",".join(value) # [ "e1", "e2", "e3" ] -> "e1,e2,e3" + return handler(value) # type: ignore[no-any-return] # actually return `list[str]` class BaseConnectorSettings(BaseConfigModel, ABC): @@ -328,6 +232,34 @@ def make_schema_generator( mode=mode, ) + @classmethod + def _extract_base_config_model_type(cls, annotation: Any) -> Any: + """Extract `BaseConfigModel` type from a field's annotation. + + Args: + annotation: The field's annotation to extract from. + + Returns: + The extracted `BaseConfigModel` type if present, otherwise `None`. + """ + # Handle `field_name: BaseConfigModel` annotations + if isinstance(annotation, type) and issubclass(annotation, BaseConfigModel): + return annotation + + # Handle `field_name: BaseConfigModel | None` / `Optional[BaseConfigModel]` annotations + annotation_origin = get_origin(annotation) + if annotation_origin in (Union, UnionType): + base_config_model_type = next( + ( + arg + for arg in get_args(annotation) + if isinstance(arg, type) and issubclass(arg, BaseConfigModel) + ), + None, + ) + if base_config_model_type: + return base_config_model_type + @classmethod def _migrate_deprecated_namespaces(cls, data: dict[str, Any]) -> dict[str, Any]: """Migrate deprecated namespaces in the configuration data. @@ -339,9 +271,8 @@ def _migrate_deprecated_namespaces(cls, data: dict[str, Any]) -> dict[str, Any]: Migrated configuration data. """ for field_name, field in cls._model_deprecated_fields.items(): - annotation = field.annotation - is_namespace = isinstance(annotation, type) and issubclass( - annotation, BaseConfigModel + is_namespace = ( + cls._extract_base_config_model_type(field.annotation) is not None ) deprecate_metadata = next( m for m in field.metadata if isinstance(m, Deprecate) @@ -382,39 +313,40 @@ def _migrate_deprecated_variables(cls, data: dict[str, Any]) -> dict[str, Any]: Migrated configuration data. """ for field_name, field in cls.model_fields.items(): - annotation = field.annotation - is_namespace = isinstance(annotation, type) and issubclass( - annotation, BaseConfigModel + base_config_model_type = cls._extract_base_config_model_type( + field.annotation ) - if is_namespace: - for ( - sub_field_name, - sub_field, - ) in annotation._model_deprecated_fields.items(): # type: ignore[union-attr] - deprecate_metadata = next( - m for m in sub_field.metadata if isinstance(m, Deprecate) - ) - new_namespace = deprecate_metadata.new_namespace - new_namespaced_var = deprecate_metadata.new_namespaced_var - new_value_factory = deprecate_metadata.new_value_factory - removal_date = deprecate_metadata.removal_date - - if new_namespaced_var: - if not isinstance(new_namespaced_var, str): - raise ValueError( - f"`new_namespaced_var` for field {sub_field_name} must be a string." - ) - - migrate_deprecated_variable( - data, - old_name=sub_field_name, - new_name=new_namespaced_var, - current_namespace=field_name, - new_namespace=new_namespace, - new_value_factory=new_value_factory, - removal_date=removal_date, + if not base_config_model_type: + continue # not a namespace, skip + + for ( + sub_field_name, + sub_field, + ) in base_config_model_type._model_deprecated_fields.items(): + deprecate_metadata = next( + m for m in sub_field.metadata if isinstance(m, Deprecate) + ) + new_namespace = deprecate_metadata.new_namespace + new_namespaced_var = deprecate_metadata.new_namespaced_var + new_value_factory = deprecate_metadata.new_value_factory + removal_date = deprecate_metadata.removal_date + + if new_namespaced_var: + if not isinstance(new_namespaced_var, str): + raise ValueError( + f"`new_namespaced_var` for field {sub_field_name} must be a string." ) + migrate_deprecated_variable( + data, + old_name=sub_field_name, + new_name=new_namespaced_var, + current_namespace=field_name, + new_namespace=new_namespace, + new_value_factory=new_value_factory, + removal_date=removal_date, + ) + return data @model_validator(mode="wrap") @@ -472,10 +404,6 @@ def to_helper_config(self) -> dict[str, Any]: return self.model_dump( mode="json", context={"mode": "pycti"}, - # Deprecated fields can be set to `None` despite their type (due to `Deprecate` annotation). - # To avoid `PydanticSerializationError`, we exclude all fields set to `None` during serialization. - # OpenCTIConnectorHelper handles missing fields with default values or internal logic. - exclude_none=True, ) diff --git a/connectors-sdk/tests/test_settings/conftest.py b/connectors-sdk/tests/test_settings/conftest.py index 42e4cd3371a..9aa0c962255 100644 --- a/connectors-sdk/tests/test_settings/conftest.py +++ b/connectors-sdk/tests/test_settings/conftest.py @@ -22,7 +22,7 @@ def mock_environment(monkeypatch): monkeypatch.setenv("OPENCTI_TOKEN", "changeme") monkeypatch.setenv("CONNECTOR_ID", "connector-poc--uid") monkeypatch.setenv("CONNECTOR_NAME", "Test Connector") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") monkeypatch.setenv("CONNECTOR_LOG_LEVEL", "error") @@ -35,7 +35,7 @@ def get_config_yml_file_path(): return Path(__file__).parent / "data" / "config.test.yml" monkeypatch.setattr( - "connectors_sdk.settings.base_settings._SettingsLoader._get_config_yml_file_path", + "connectors_sdk.settings._settings_loader._SettingsLoader._get_config_yml_file_path", get_config_yml_file_path, ) @@ -48,6 +48,6 @@ def get_dot_env_file_path(): return Path(__file__).parent / "data" / ".env.test" monkeypatch.setattr( - "connectors_sdk.settings.base_settings._SettingsLoader._get_dot_env_file_path", + "connectors_sdk.settings._settings_loader._SettingsLoader._get_dot_env_file_path", get_dot_env_file_path, ) diff --git a/connectors-sdk/tests/test_settings/data/.env.test b/connectors-sdk/tests/test_settings/data/.env.test index bf5927c7dbe..8c426344946 100644 --- a/connectors-sdk/tests/test_settings/data/.env.test +++ b/connectors-sdk/tests/test_settings/data/.env.test @@ -2,6 +2,6 @@ OPENCTI_URL=http://localhost:8080 OPENCTI_TOKEN=changeme CONNECTOR_ID=connector-poc--uid CONNECTOR_NAME=Test Connector -CONNECTOR_SCOPE=test -CONNECTOR_LOG_LEVEL=error +CONNECTOR_SCOPE=scope1,scope2 +CONNECTOR_LOG_LEVEL=debug CONNECTOR_DURATION_PERIOD=PT5M \ No newline at end of file diff --git a/connectors-sdk/tests/test_settings/data/config.test.yml b/connectors-sdk/tests/test_settings/data/config.test.yml index 730f7b2787c..bea9f9d5014 100644 --- a/connectors-sdk/tests/test_settings/data/config.test.yml +++ b/connectors-sdk/tests/test_settings/data/config.test.yml @@ -5,7 +5,7 @@ opencti: connector: id: connector-poc--uid name: Test Connector - scope: test - log_level: error + scope: scope1,scope2 + log_level: debug duration_period: PT5M \ No newline at end of file diff --git a/connectors-sdk/tests/test_settings/test_annotated_types.py b/connectors-sdk/tests/test_settings/test_annotated_types.py index bc611dca536..620205271d2 100644 --- a/connectors-sdk/tests/test_settings/test_annotated_types.py +++ b/connectors-sdk/tests/test_settings/test_annotated_types.py @@ -1,5 +1,4 @@ from datetime import datetime, timedelta, timezone -from types import SimpleNamespace import freezegun import pytest @@ -8,7 +7,6 @@ ListFromString, parse_comma_separated_list, parse_iso_string, - serialize_list_of_strings, ) from pydantic import TypeAdapter @@ -45,20 +43,6 @@ def test_parse_comma_separated_list_passthrough() -> None: assert parse_comma_separated_list(["a", "b"]) == ["a", "b"] -def test_serialize_list_of_strings_handles_pycti_mode() -> None: - info = SimpleNamespace(context={"mode": "pycti"}) - assert serialize_list_of_strings(["a", "b"], info) == "a,b" - - -@pytest.mark.parametrize("context", [None, {}, {"mode": "other"}]) -def test_serialize_list_of_strings_handles_non_pycti_modes( - context: dict[str, str] | None, -) -> None: - info = SimpleNamespace(context=context) - value = ["a", "b"] - assert serialize_list_of_strings(value, info) == value - - def test_list_from_string_accepts_string_input() -> None: value = TypeAdapter(ListFromString).validate_python("a,b,c") assert value == ["a", "b", "c"] @@ -74,30 +58,6 @@ def test_list_from_string_dumps_valid_json() -> None: assert value == ["a", "b"] -@pytest.mark.parametrize( - "input,expected", - [ - pytest.param( - ["a", "b", "c"], - "a,b,c", - id="list_of_strings", - ), - pytest.param( - [], - "", - id="empty_list", - ), # empty list -> empty string - ], -) -def test_list_from_string_dumps_valid_json_in_pycti_mode( - input: list[str], expected: str -) -> None: - value = TypeAdapter(ListFromString).dump_python( - input, mode="json", context={"mode": "pycti"} - ) - assert value == expected - - # DatetimeFromIsoString diff --git a/connectors-sdk/tests/test_settings/test_base_settings.py b/connectors-sdk/tests/test_settings/test_base_settings.py index 2d8f3f9db9c..a6429289812 100644 --- a/connectors-sdk/tests/test_settings/test_base_settings.py +++ b/connectors-sdk/tests/test_settings/test_base_settings.py @@ -4,14 +4,14 @@ from unittest.mock import patch import pytest +from connectors_sdk.settings._settings_loader import _SettingsLoader from connectors_sdk.settings.base_settings import ( BaseConfigModel, BaseConnectorSettings, - _SettingsLoader, ) from connectors_sdk.settings.deprecations import Deprecate, DeprecatedField from connectors_sdk.settings.exceptions import ConfigValidationError -from pydantic import Field, HttpUrl +from pydantic import Field, HttpUrl, SecretStr def test_base_config_model_should_retrieve_deprecated_fields(): @@ -48,12 +48,24 @@ class TestConfig(BaseConfigModel): assert "old_field" in TestConfig._model_deprecated_fields +def test_base_config_model_should_make_deprecated_fields_optional(): + """Test that `BaseConfigModel` subclasses set `default` to `None` for deprecated fields.""" + + # Given: A deprecated field explicitly defined as required (non-optional) + class TestConfig(BaseConfigModel): + old_field: str = DeprecatedField() # type should be overwritten to `str | None` + + # When: The model field definitions are built + # Then: Deprecated field annotation is normalized to `str | None` to make it optional + assert TestConfig.model_fields["old_field"].annotation == str | None + assert TestConfig._model_deprecated_fields["old_field"].annotation == str | None + + def test_base_config_model_should_set_default_to_none_for_deprecated_fields(): """Test that `BaseConfigModel` subclasses set `default` to `None` for deprecated fields.""" # Given: A deprecated field explicitly defines a non-None default class TestConfig(BaseConfigModel): - test_field: str = Field(default="test") old_field: str = DeprecatedField( default="deprecated default" # should be overwritten to None ) @@ -208,8 +220,8 @@ def test_settings_loader_should_parse_config_yml_file(mock_config_yml_file_prese "id": "connector-poc--uid", "name": "Test Connector", "duration_period": "PT5M", - "log_level": "error", - "scope": "test", + "log_level": "debug", + "scope": "scope1,scope2", }, } @@ -232,8 +244,8 @@ def test_settings_loader_should_parse_dot_env_file(mock_dot_env_file_presence): "connector_id": "connector-poc--uid", "connector_name": "Test Connector", "connector_duration_period": "PT5M", - "connector_log_level": "error", - "connector_scope": "test", + "connector_log_level": "debug", + "connector_scope": "scope1,scope2", } @@ -271,8 +283,8 @@ def test_settings_loader_should_parse_config_yml_from_model( assert settings_dict["opencti"]["token"] == "changeme" assert settings_dict["connector"]["id"] == "connector-poc--uid" assert settings_dict["connector"]["name"] == "Test Connector" - assert settings_dict["connector"]["scope"] == "test" - assert settings_dict["connector"]["log_level"] == "error" + assert settings_dict["connector"]["scope"] == "scope1,scope2" + assert settings_dict["connector"]["log_level"] == "debug" def test_settings_loader_should_parse_dot_env_from_model(mock_dot_env_file_presence): @@ -292,8 +304,8 @@ def test_settings_loader_should_parse_dot_env_from_model(mock_dot_env_file_prese assert settings_dict["opencti"]["token"] == "changeme" assert settings_dict["connector"]["id"] == "connector-poc--uid" assert settings_dict["connector"]["name"] == "Test Connector" - assert settings_dict["connector"]["scope"] == "test" - assert settings_dict["connector"]["log_level"] == "error" + assert settings_dict["connector"]["scope"] == "scope1,scope2" + assert settings_dict["connector"]["log_level"] == "debug" def test_settings_loader_should_parse_os_environ_from_model(mock_environment): @@ -313,71 +325,71 @@ def test_settings_loader_should_parse_os_environ_from_model(mock_environment): assert settings_dict["opencti"]["token"] == "changeme" assert settings_dict["connector"]["id"] == "connector-poc--uid" assert settings_dict["connector"]["name"] == "Test Connector" - assert settings_dict["connector"]["scope"] == "test" + assert settings_dict["connector"]["scope"] == "scope1,scope2" assert settings_dict["connector"]["log_level"] == "error" -def test_base_connector_settings_should_validate_settings_from_config_yaml_file( - mock_config_yml_file_presence, +def test_base_connector_settings_should_validate_settings_from_dot_env_file( + mock_dot_env_file_presence, ): """ - Test that `BaseConnectorSettings` casts and validates config vars in `config.yml`. - For testing purpose, the path of `config.yml` file is `tests/test_settings/data/config.test.yml`. + Test that `BaseConnectorSettings` casts and validates env vars in `.env`. + For testing purpose, the path of `.env` file is `tests/test_settings/data/.env.test`. """ - # Given: Valid connector settings are provided through config.yml fixture + # Given: Valid connector settings are provided through .env fixture # When: BaseConnectorSettings is instantiated settings = BaseConnectorSettings() # Then: Values are validated and cast to expected runtime types assert settings.opencti.url == HttpUrl("http://localhost:8080/") - assert settings.opencti.token == "changeme" + assert settings.opencti.token == SecretStr("changeme") assert settings.connector.id == "connector-poc--uid" assert settings.connector.name == "Test Connector" - assert settings.connector.scope == ["test"] - assert settings.connector.log_level == "error" + assert settings.connector.scope == ["scope1", "scope2"] + assert settings.connector.log_level == "debug" -def test_base_connector_settings_should_validate_settings_from_dot_env_file( - mock_dot_env_file_presence, +def test_base_connector_settings_should_validate_settings_from_os_environ( + mock_environment, ): """ - Test that `BaseConnectorSettings` casts and validates env vars in `.env`. - For testing purpose, the path of `.env` file is `tests/test_settings/data/.env.test`. + Test that `BaseConnectorSettings` casts and validates env vars in `os.environ`. + For testing purpose, `os.environ` is patched. """ - # Given: Valid connector settings are provided through .env fixture + # Given: Valid connector settings are provided through patched os.environ # When: BaseConnectorSettings is instantiated settings = BaseConnectorSettings() # Then: Values are validated and cast to expected runtime types assert settings.opencti.url == HttpUrl("http://localhost:8080/") - assert settings.opencti.token == "changeme" + assert settings.opencti.token == SecretStr("changeme") assert settings.connector.id == "connector-poc--uid" assert settings.connector.name == "Test Connector" - assert settings.connector.scope == ["test"] + assert settings.connector.scope == ["scope1", "scope2"] assert settings.connector.log_level == "error" -def test_base_connector_settings_should_validate_settings_from_os_environ( - mock_environment, +def test_base_connector_settings_should_validate_settings_from_config_yaml_file( + mock_config_yml_file_presence, ): """ - Test that `BaseConnectorSettings` casts and validates env vars in `os.environ`. - For testing purpose, `os.environ` is patched. + Test that `BaseConnectorSettings` casts and validates config vars in `config.yml`. + For testing purpose, the path of `config.yml` file is `tests/test_settings/data/config.test.yml`. """ - # Given: Valid connector settings are provided through patched os.environ + # Given: Valid connector settings are provided through config.yml fixture # When: BaseConnectorSettings is instantiated settings = BaseConnectorSettings() # Then: Values are validated and cast to expected runtime types assert settings.opencti.url == HttpUrl("http://localhost:8080/") - assert settings.opencti.token == "changeme" + assert settings.opencti.token == SecretStr("changeme") assert settings.connector.id == "connector-poc--uid" assert settings.connector.name == "Test Connector" - assert settings.connector.scope == ["test"] - assert settings.connector.log_level == "error" + assert settings.connector.scope == ["scope1", "scope2"] + assert settings.connector.log_level == "debug" def test_base_connector_settings_should_raise_when_missing_mandatory_env_vars(): @@ -398,20 +410,25 @@ def test_base_connector_settings_should_provide_helper_config(mock_environment): # Given: A valid BaseConnectorSettings instance built from patched environment # When: OpenCTIConnectorHelper config dict is generated settings = BaseConnectorSettings() + json_dump = settings.model_dump(mode="json") opencti_dict = settings.to_helper_config() + # Then: The regular JSON dump of settings does not expose the secret token value + assert json_dump["opencti"]["token"] == "**********" + assert json_dump["connector"]["scope"] == ["scope1", "scope2"] + # Then: The resulting helper config dict matches expected structure and values assert opencti_dict == { + "opencti": { + "token": "changeme", # clear token + "url": "http://localhost:8080/", + }, "connector": { "duration_period": "PT5M", "id": "connector-poc--uid", "log_level": "error", "name": "Test Connector", - "scope": "test", - }, - "opencti": { - "token": "changeme", - "url": "http://localhost:8080/", + "scope": "scope1,scope2", # comma-separated string }, } diff --git a/connectors-sdk/tests/test_settings/test_deprecation_migration.py b/connectors-sdk/tests/test_settings/test_deprecation_migration.py index 1a28a51149e..a6188ba49b3 100644 --- a/connectors-sdk/tests/test_settings/test_deprecation_migration.py +++ b/connectors-sdk/tests/test_settings/test_deprecation_migration.py @@ -36,7 +36,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("CONNECTOR_ID", "test-id") monkeypatch.setenv("CONNECTOR_NAME", "Test") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") monkeypatch.setenv("CONNECTOR_OLD_FIELD", "old_value") @@ -73,7 +73,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("CONNECTOR_ID", "test-id") monkeypatch.setenv("CONNECTOR_NAME", "Test") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") # When: Settings initialization evaluates migration metadata @@ -108,7 +108,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("CONNECTOR_ID", "test-id") monkeypatch.setenv("CONNECTOR_NAME", "Test") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") monkeypatch.setenv("CONNECTOR_OLD_VALUE", "5") @@ -139,7 +139,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("CONNECTOR_ID", "test-id") monkeypatch.setenv("CONNECTOR_NAME", "Test") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") # When: TestSettings initialization evaluates namespace migration metadata @@ -171,7 +171,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("OLD_CONNECTOR_ID", "test-id") monkeypatch.setenv("OLD_CONNECTOR_NAME", "Test") - monkeypatch.setenv("OLD_CONNECTOR_SCOPE", "test") + monkeypatch.setenv("OLD_CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("OLD_CONNECTOR_DURATION_PERIOD", "PT5M") # When: TestSettings initialization validates deprecated namespace migration @@ -203,7 +203,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("CONNECTOR_ID", "test-id") monkeypatch.setenv("CONNECTOR_NAME", "Test") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") monkeypatch.setenv("CONNECTOR_SPECIAL_FIELD", "special_value") @@ -237,7 +237,7 @@ class TestSettings(BaseConnectorSettings): monkeypatch.setenv("OPENCTI_TOKEN", "test-token") monkeypatch.setenv("CONNECTOR_ID", "test-id") monkeypatch.setenv("CONNECTOR_NAME", "Test") - monkeypatch.setenv("CONNECTOR_SCOPE", "test") + monkeypatch.setenv("CONNECTOR_SCOPE", "scope1,scope2") monkeypatch.setenv("CONNECTOR_DURATION_PERIOD", "PT5M") monkeypatch.setenv("CONNECTOR_OLD_FIELD", "migrated_value") diff --git a/connectors-sdk/tests/test_settings/test_settings_loader.py b/connectors-sdk/tests/test_settings/test_settings_loader.py new file mode 100644 index 00000000000..1d4a0d038ae --- /dev/null +++ b/connectors-sdk/tests/test_settings/test_settings_loader.py @@ -0,0 +1,268 @@ +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest +from connectors_sdk.settings._settings_loader import _SettingsLoader +from connectors_sdk.settings.base_settings import ( + BaseConfigModel, + BaseConnectorSettings, +) +from connectors_sdk.settings.deprecations import DeprecatedField +from pydantic import Field + + +def test_settings_loader_should_get_connector_main_path(mock_main_path): + """ + Test that `_SettingsLoader._get_connector_main_path` locates connector's `main.py`. + For testing purpose, a fake path is assigned to `sys.modules[__main__].__file__`. + """ + + # Given: The connector main module path is available + # When: The main path resolver is executed + main_path = _SettingsLoader._get_connector_main_path() + + # Then: The resolved main.py path matches the expected connector location + assert main_path == Path("/app/src/main.py").resolve() + + +def test_settings_loader_should_raise_when_main_module_misses_file_attribute( + mock_main_path, +): + """ + Test that `_SettingsLoader._get_connector_main_path` raises a meaningful error in case `__main__.__file__` is missing. + For testing purpose, `sys.modules[__main__].__file__` is set to `None`. + """ + + # Given: The __main__.__file__ attribute is missing + sys.modules["__main__"].__file__ = None + + # When: The main path resolver is executed + # Then: A runtime error is raised to signal invalid execution context + with pytest.raises(RuntimeError): + _SettingsLoader._get_connector_main_path() + + +def test_settings_loader_should_get_legacy_config_yml_file_path( + mock_main_path, +): + """ + Test that `_SettingsLoader._get_config_yml_file_path` locates connector's `config.yml` (legacy path). + For testing purpose, a fake path is assigned to `sys.modules[__main__].__file__`. + """ + + def is_file(self: Path) -> bool: + return self.name == "config.yml" + + # Given: Legacy config file (/src/config.yml) is present + with patch("pathlib.Path.is_file", is_file): + # When: The config.yml path resolver is executed + config_yml_file_path = _SettingsLoader._get_config_yml_file_path() + + # Then: The legacy config.yml path is returned + assert config_yml_file_path == Path("/app/src/config.yml").resolve() + + +def test_settings_loader_should_get_config_yml_file_path(mock_main_path): + """ + Test that `_SettingsLoader._get_config_yml_file_path` locates connector's `config.yml` (new path). + For testing purpose, a fake path is assigned to `sys.modules[__main__].__file__`. + """ + + def is_file(self: Path) -> bool: + return self.name == "config.yml" and self.parent.name != "src" + + # Given: Root config file (/config.yml) is present + with patch("pathlib.Path.is_file", is_file): + # When: The config.yml path resolver is executed + config_yml_file_path = _SettingsLoader._get_config_yml_file_path() + + # Then: The new config.yml path is returned + assert config_yml_file_path == Path("/app/config.yml").resolve() + + +def test_settings_loader_should_get_dot_env_file_path(mock_main_path): + """ + Test that `_SettingsLoader._get_dot_env_file_path` locates connector's `.env`. + For testing purpose, a fake path is assigned to `sys.modules[__main__].__file__`. + """ + + def is_file(self: Path) -> bool: + return self.name == ".env" + + # Given: Root env file (/.env) is present + with patch("pathlib.Path.is_file", is_file): + # When: The .env path resolver is executed + dot_env_file_path = _SettingsLoader._get_dot_env_file_path() + + # Then: The .env file path is returned + assert dot_env_file_path == Path("/app/.env").resolve() + + +def test_settings_loader_should_parse_config_yml_file(mock_config_yml_file_presence): + """ + Test that `_SettingsLoader()` parses config vars in `config.yml`. + For testing purpose, the path of `config.yml` file is `tests/test_settings/data/config.test.yml`. + """ + # Given: A valid config.yml + # When: The settings loader is instantiated and dumped + settings_loader = _SettingsLoader() + settings_dict = settings_loader.model_dump() + + # Then: Parsed nested settings match expected config.yml values + assert settings_dict == { + "opencti": { + "url": "http://localhost:8080", + "token": "changeme", + }, + "connector": { + "id": "connector-poc--uid", + "name": "Test Connector", + "duration_period": "PT5M", + "log_level": "debug", + "scope": "scope1,scope2", + }, + } + + +def test_settings_loader_should_parse_dot_env_file(mock_dot_env_file_presence): + """ + Test that `_SettingsLoader()` parses env vars in `.env`. + For testing purpose, the path of `.env` file is `tests/test_settings/data/.env.test`. + """ + + # Given: A valid .env file + # When: The settings loader is instantiated and dumped + settings_loader = _SettingsLoader() + settings_dict = settings_loader.model_dump() + + # Then: Parsed flat settings match expected environment variables + assert settings_dict == { + "opencti_url": "http://localhost:8080", + "opencti_token": "changeme", + "connector_id": "connector-poc--uid", + "connector_name": "Test Connector", + "connector_duration_period": "PT5M", + "connector_log_level": "debug", + "connector_scope": "scope1,scope2", + } + + +def test_settings_loader_should_not_parse_os_environ(mock_environment): + """ + Test that `_SettingsLoader()` does not parse env vars from `os.environ` (for security purposes). + For testing purpose, `os.environ` is patched. + """ + + # Given: Valid environment variables + # When: The settings loader is instantiated and dumped + settings_loader = _SettingsLoader() + settings_dict = settings_loader.model_dump() + + # Then: No implicit values are parsed + assert settings_dict == {} + + +def test_settings_loader_should_parse_config_yml_from_model( + mock_config_yml_file_presence, +): + """ + Test that `_SettingsLoader.build_loader_from_model` returns a `BaseSettings` subclass + capable of parsing `config.yml` according to the given `BaseModel`. + For testing purpose, the path of `config.yml` file is `tests/test_settings/data/config.test.yml`. + """ + + # Given: A model-aware loader is built for BaseConnectorSettings with config.yml fixture + # When: The loader instance parses and dumps values + settings_loader = _SettingsLoader.build_loader_from_model(BaseConnectorSettings) + settings_dict = settings_loader().model_dump() + + # Then: Parsed nested settings expose expected OpenCTI and connector values + assert settings_dict["opencti"]["url"] == "http://localhost:8080" + assert settings_dict["opencti"]["token"] == "changeme" + assert settings_dict["connector"]["id"] == "connector-poc--uid" + assert settings_dict["connector"]["name"] == "Test Connector" + assert settings_dict["connector"]["scope"] == "scope1,scope2" + assert settings_dict["connector"]["log_level"] == "debug" + + +def test_settings_loader_should_parse_dot_env_from_model(mock_dot_env_file_presence): + """ + Test that `_SettingsLoader.build_loader_from_model` returns a `BaseSettings` subclass + capable of parsing `.env` according to the given `BaseModel`. + For testing purpose, the path of `.env` file is `tests/test_settings/data/.env.test`. + """ + + # Given: A model-aware loader is built for BaseConnectorSettings with .env fixture + # When: The loader instance parses and dumps values + settings_loader = _SettingsLoader.build_loader_from_model(BaseConnectorSettings) + settings_dict = settings_loader().model_dump() + + # Then: Parsed nested settings expose expected OpenCTI and connector values + assert settings_dict["opencti"]["url"] == "http://localhost:8080" + assert settings_dict["opencti"]["token"] == "changeme" + assert settings_dict["connector"]["id"] == "connector-poc--uid" + assert settings_dict["connector"]["name"] == "Test Connector" + assert settings_dict["connector"]["scope"] == "scope1,scope2" + assert settings_dict["connector"]["log_level"] == "debug" + + +def test_settings_loader_should_parse_os_environ_from_model(mock_environment): + """ + Test that `_SettingsLoader.build_loader_from_model` returns a `BaseSettings` subclass + capable of parsing `os.environ` according to the given `BaseModel`. + For testing purpose, `os.environ` is patched. + """ + + # Given: A model-aware loader is built for BaseConnectorSettings with patched os.environ + # When: The loader instance parses and dumps values + settings_loader = _SettingsLoader.build_loader_from_model(BaseConnectorSettings) + settings_dict = settings_loader().model_dump() + + # Then: Parsed nested settings expose expected OpenCTI and connector values + assert settings_dict["opencti"]["url"] == "http://localhost:8080" + assert settings_dict["opencti"]["token"] == "changeme" + assert settings_dict["connector"]["id"] == "connector-poc--uid" + assert settings_dict["connector"]["name"] == "Test Connector" + assert settings_dict["connector"]["scope"] == "scope1,scope2" + assert settings_dict["connector"]["log_level"] == "error" + + +def test_settings_loader_should_parse_os_environ_from_model_with_deprecated_fields( + monkeypatch, mock_environment +): + """ + Test that `_SettingsLoader.build_loader_from_model` returns a `BaseSettings` subclass + capable of parsing `os.environ` according to the given `BaseModel` with deprecated fields. + For testing purpose, `os.environ` is patched. + """ + + monkeypatch.setenv("DEPRECATED_NAMESPACE_TEST_FIELD", "deprecated_value") + + # Given: Connector settings model with deprecated fields + class DeprecatedConfig(BaseConfigModel): + test_field: str = Field( + description="This is a test field.", + ) + + class DeprecatedConnectorSettings(BaseConnectorSettings): + """A connector settings model with deprecated namespaces.""" + + deprecated_namespace: DeprecatedConfig = DeprecatedField( + deprecated="This namespace is deprecated.", + ) + + # When: The settings loader instance parses and dumps values + settings_loader = _SettingsLoader.build_loader_from_model( + DeprecatedConnectorSettings + ) + settings_dict = settings_loader().model_dump() + + # Then: Parsed nested settings expose expected OpenCTI and connector values + assert settings_dict["opencti"]["url"] == "http://localhost:8080" + assert settings_dict["opencti"]["token"] == "changeme" + assert settings_dict["connector"]["id"] == "connector-poc--uid" + assert settings_dict["connector"]["name"] == "Test Connector" + assert settings_dict["connector"]["scope"] == "scope1,scope2" + assert settings_dict["connector"]["log_level"] == "error" + assert settings_dict["deprecated_namespace"]["test_field"] == "deprecated_value" diff --git a/external-import/ctm360-cyberblindspot-feed/tests/test_connector/test_settings.py b/external-import/ctm360-cyberblindspot-feed/tests/test_connector/test_settings.py index 5518d96c50f..5cdbae4acd9 100644 --- a/external-import/ctm360-cyberblindspot-feed/tests/test_connector/test_settings.py +++ b/external-import/ctm360-cyberblindspot-feed/tests/test_connector/test_settings.py @@ -24,7 +24,8 @@ def test_opencti_token(self): """OpenCTI token should match the environment variable.""" settings = ConnectorSettings() assert ( - settings.opencti.token == "test-token-00000000-0000-0000-0000-000000000000" + settings.opencti.token.get_secret_value() + == "test-token-00000000-0000-0000-0000-000000000000" ) def test_connector_id(self): diff --git a/external-import/ctm360-cyna-feed/tests/test_connector/test_settings.py b/external-import/ctm360-cyna-feed/tests/test_connector/test_settings.py index 14dc46d0006..42a706e91ce 100644 --- a/external-import/ctm360-cyna-feed/tests/test_connector/test_settings.py +++ b/external-import/ctm360-cyna-feed/tests/test_connector/test_settings.py @@ -24,7 +24,8 @@ def test_opencti_token(self): """OpenCTI token should match the environment variable.""" settings = ConnectorSettings() assert ( - settings.opencti.token == "test-token-00000000-0000-0000-0000-000000000000" + settings.opencti.token.get_secret_value() + == "test-token-00000000-0000-0000-0000-000000000000" ) def test_connector_id(self): diff --git a/external-import/ctm360-hackerview-feed/tests/test_connector/test_settings.py b/external-import/ctm360-hackerview-feed/tests/test_connector/test_settings.py index 32f51957923..16ccd5f1285 100644 --- a/external-import/ctm360-hackerview-feed/tests/test_connector/test_settings.py +++ b/external-import/ctm360-hackerview-feed/tests/test_connector/test_settings.py @@ -28,7 +28,8 @@ def test_opencti_token(self): """OpenCTI token should match the environment variable.""" settings = ConnectorSettings() assert ( - settings.opencti.token == "test-token-00000000-0000-0000-0000-000000000000" + settings.opencti.token.get_secret_value() + == "test-token-00000000-0000-0000-0000-000000000000" ) def test_connector_id(self): diff --git a/external-import/flare/tests/test_connector/test_settings.py b/external-import/flare/tests/test_connector/test_settings.py index 11367222d52..14a7ea89cb6 100644 --- a/external-import/flare/tests/test_connector/test_settings.py +++ b/external-import/flare/tests/test_connector/test_settings.py @@ -102,7 +102,7 @@ def test_to_helper_config_structure(self, required_env: None) -> None: settings = ConnectorSettings() config = settings.to_helper_config() assert config["opencti"]["url"] == str(settings.opencti.url) - assert config["opencti"]["token"] == settings.opencti.token + assert config["opencti"]["token"] == settings.opencti.token.get_secret_value() assert config["connector"]["id"] == settings.connector.id assert config["connector"]["type"] == "EXTERNAL_IMPORT" assert config["connector"]["name"] == settings.connector.name diff --git a/external-import/ransomlook/__metadata__/connector_config_schema.json b/external-import/ransomlook/__metadata__/connector_config_schema.json index fa8bb484503..8a1ab739590 100644 --- a/external-import/ransomlook/__metadata__/connector_config_schema.json +++ b/external-import/ransomlook/__metadata__/connector_config_schema.json @@ -12,7 +12,9 @@ }, "OPENCTI_TOKEN": { "description": "The API token to connect to OpenCTI.", - "type": "string" + "format": "password", + "type": "string", + "writeOnly": true }, "CONNECTOR_NAME": { "default": "RansomLook", @@ -332,4 +334,4 @@ "OPENCTI_TOKEN" ], "additionalProperties": false -} \ No newline at end of file +} diff --git a/external-import/socprime/tests/socprime/test_config.py b/external-import/socprime/tests/socprime/test_config.py index 138e76cfea3..03b88e4df29 100644 --- a/external-import/socprime/tests/socprime/test_config.py +++ b/external-import/socprime/tests/socprime/test_config.py @@ -1,7 +1,7 @@ from datetime import timedelta import pytest -from pydantic import HttpUrl +from pydantic import HttpUrl, SecretStr from socprime import SocprimeConnector from socprime.settings import ConnectorSettings @@ -29,7 +29,7 @@ def test_config_settings() -> None: config = ConnectorSettings().model_dump() assert config["opencti"]["url"] == HttpUrl("http://test-opencti-url/") - assert config["opencti"]["token"] == "test-opencti-token" + assert config["opencti"]["token"] == SecretStr("test-opencti-token") assert config["connector"]["id"] == "test-connector-id" assert config["connector"]["name"] == "Soc Prime" diff --git a/internal-enrichment/censys-enrichment/tests/censys_enrichment/test_config.py b/internal-enrichment/censys-enrichment/tests/censys_enrichment/test_config.py index 46a1b1e3126..40d87f264c1 100644 --- a/internal-enrichment/censys-enrichment/tests/censys_enrichment/test_config.py +++ b/internal-enrichment/censys-enrichment/tests/censys_enrichment/test_config.py @@ -10,7 +10,7 @@ def test_config() -> None: # Test config from env assert config.opencti.url == HttpUrl("http://test") - assert config.opencti.token == "opencti-token" + assert config.opencti.token.get_secret_value() == "opencti-token" assert ( config.censys_enrichment.organisation_id.get_secret_value() diff --git a/internal-enrichment/polyswarm-sandbox/tests/test_connector.py b/internal-enrichment/polyswarm-sandbox/tests/test_connector.py index 3f681718afe..379bea3991d 100644 --- a/internal-enrichment/polyswarm-sandbox/tests/test_connector.py +++ b/internal-enrichment/polyswarm-sandbox/tests/test_connector.py @@ -131,7 +131,7 @@ def test_config_loads_with_env_vars(self, monkeypatch): config = ConnectorSettings() assert str(config.opencti.url).rstrip("/") == "http://localhost:8080" - assert config.opencti.token == "test-token" + assert config.opencti.token.get_secret_value() == "test-token" assert config.polyswarm.api_key.get_secret_value() == "test-api-key" def test_config_defaults(self, monkeypatch): diff --git a/internal-enrichment/qualys-cve-enrichment/tests/test_connector/test_settings.py b/internal-enrichment/qualys-cve-enrichment/tests/test_connector/test_settings.py index a59c09f4967..4bc06d093fe 100644 --- a/internal-enrichment/qualys-cve-enrichment/tests/test_connector/test_settings.py +++ b/internal-enrichment/qualys-cve-enrichment/tests/test_connector/test_settings.py @@ -28,7 +28,8 @@ def test_opencti_token(self): """OpenCTI token should match the environment variable.""" settings = ConnectorSettings() assert ( - settings.opencti.token == "test-token-00000000-0000-0000-0000-000000000000" + settings.opencti.token.get_secret_value() + == "test-token-00000000-0000-0000-0000-000000000000" ) def test_connector_id(self): diff --git a/stream/microsoft-sentinel-intel/tests/microsoft_sentinel_intel/test_config.py b/stream/microsoft-sentinel-intel/tests/microsoft_sentinel_intel/test_config.py index f66744d3c0d..1c248327bc5 100644 --- a/stream/microsoft-sentinel-intel/tests/microsoft_sentinel_intel/test_config.py +++ b/stream/microsoft-sentinel-intel/tests/microsoft_sentinel_intel/test_config.py @@ -1,6 +1,6 @@ import pytest from microsoft_sentinel_intel.settings import ConnectorSettings -from pydantic import HttpUrl +from pydantic import HttpUrl, SecretStr @pytest.mark.usefixtures("mock_microsoft_sentinel_intel_config") @@ -8,7 +8,7 @@ def test_config() -> None: config = ConnectorSettings().model_dump() assert config["opencti"]["url"] == HttpUrl("http://test-opencti-url/") - assert config["opencti"]["token"] == "test-opencti-token" + assert config["opencti"]["token"] == SecretStr("test-opencti-token") assert config["connector"]["id"] == "test-connector-id" assert config["connector"]["name"] == "External Import Connector Template"