Add security config validation based on configuration provider (#22226) (#23000)

* log source and provider and call pdb debugger

* remove pdb call and source field

* log both source and provider

* log provider only

* inject provider as property

* add security module to read integration security agent configs

* add secure_field property in the spec and the models

* update SecurityConfig module, and improve security_field validation logic and model generation

* rename secure_field to require_trusted_provider

* improve models generation to be more concise

* mark http and jmx filepath properties as protected

* improve model generation to be more concise

* generate new models

* improve tests and fix security config default allowlist behavior

* add fallback security validation in the base check

* ddev validate config and models

* add require_trusted_provider to the set of allowed value fields

* code cleanup

* ddev validate models -s

* changelog

* Fix path traversal and sibling-directory bypass in is_file_path_allowed

Bare startswith allowed bypasses like /allowed-extra/ matching /allowed
and /allowed/../etc/passwd traversing outside. Now resolves symlinks
with os.path.realpath and enforces os.sep boundary checks.



* Make DEFAULT_TRUSTED_PROVIDERS immutable

Change from mutable list to tuple to prevent accidental mutation of
shared module-level state across SecurityConfig instances.



* Add type hints to core.py and utils.py

Per AGENTS.md guidelines, new code should include type hints using
modern syntax.



* fix tests models/spec to include tls_cert param

* Catch ValueError in GLOBAL_SECURE_FIELDS fallback

ValueError from check_field_trusted_provider was escaping the
except ValidationError handler. Add explicit ValueError catch
to wrap it as ConfigurationError.



* Remove phantom certificate_path from GLOBAL_SECURE_FIELDS

certificate_path doesn't exist in any integration spec.



* Add missing JMX fields to GLOBAL_SECURE_FIELDS

Add java_bin_path, trust_store_path, key_store_path, and
tools_jar_path for fallback coverage of JMX template fields.



* Resolve allowlist paths in is_file_path_allowed

Apply os.path.realpath() to allowlist entries so symlinks in
the allowlist are resolved before comparison.



* Remove dead branch in check_field_trusted_provider

security_config cannot be None when the error raises, since
validate_require_trusted_provider returns True for None.



* Use list[str] in model_info.py

Replace List[str] with modern list[str] syntax and remove
unused typing import.



* Add test coverage for fallback, allowlist, and excluded_checks

Cover three security-critical code paths that had zero test
coverage: GLOBAL_SECURE_FIELDS fallback blocking, allowlist
bypass, and excluded_checks bypass.



* Assert ConfigurationError instead of Exception in tests

Use specific ConfigurationError in pytest.raises to avoid
masking unexpected exceptions.



* Match ConfigurationError in exception message instead of Exception

dd_run_check wraps all errors as Exception, so match on
ConfigurationError in the traceback string to verify the
correct exception type is raised internally.



* Fix regex patterns to match multiline exception messages

Add (?s) dotall flag so ConfigurationError match works across
newlines in dd_run_check's traceback output.



* Add auth_token to test model and test for non-string secure field blocking

Add an object-typed `auth_token` field with `require_trusted_provider: true`
to the test spec/model and a test asserting that non-string secure fields
are blocked from untrusted providers.



* Enforce trusted-provider checks for non-string secure fields

Remove the `isinstance(value, str)` early return that let non-string values
bypass validation. Non-string values (e.g. object-typed auth_token) from
untrusted providers are now blocked; the allowlist escape is applied only to
string file paths.



* fix validation for non-string fields such as auth_token, add more tests

* ddev validate models -s

* remove ValueError catch from base.py

* fix fallback security validation placement

* fix license headers year from 2024 to 2026

* address review: fix a bug where empty trusted_providers list would default to default providers

* Revert all changes outside datadog_checks_base/ to origin/master

Reset integration model regenerations, changelogs, and unrelated
dependency changes so the branch only contains datadog_checks_base/ diffs.



* revert datadog_checks_base models regeneration

* address juanpe review

---------


(cherry picked from commit 31eda08)

Co-authored-by: NouemanKHAL <noueman.khalikine@datadoghq.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 3 people

datadog_checks_base/changelog.d/22226.added

@@ -0,0 +1 @@
+Add support for require_trusted_provider security validation

datadog_checks_base/datadog_checks/base/checks/base.py

@@ -27,6 +27,11 @@
 from datadog_checks.base.utils.common import ensure_bytes, to_native_string
 from datadog_checks.base.utils.fips import enable_fips
 from datadog_checks.base.utils.format import json
+from datadog_checks.base.utils.models.validation.security import (
+    DEFAULT_TRUSTED_PROVIDERS,
+    SecurityConfig,
+    check_field_trusted_provider,
+)
 from datadog_checks.base.utils.tagging import GENERIC_TAGS
 from datadog_checks.base.utils.tracing import traced_class
 
@@ -75,6 +80,27 @@
 TYPO_SIMILARITY_THRESHOLD = 0.95
 
 
+# Global list of secure fields that require trusted provider validation.
+# This provides a fallback security check for integrations that haven't
+# regenerated their models with require_trusted_provider in the spec.
+GLOBAL_SECURE_FIELDS = frozenset(
+    [
+        'tls_cert',
+        'tls_private_key',
+        'tls_ca_cert',
+        'kerberos_keytab',
+        'kerberos_cache',
+        'bearer_token_path',
+        'auth_token',
+        'private_key_path',
+        'java_bin_path',
+        'trust_store_path',
+        'key_store_path',
+        'tools_jar_path',
+    ]
+)
+
+
 @traced_class
 class AgentCheck(object):
     """
@@ -195,6 +221,7 @@ def __init__(self, *args, **kwargs):
         instance = instances[0] if instances else None
 
         self.check_id = ''
+        self.provider = ''
         self.name = name  # type: str
         self.init_config = init_config  # type: InitConfigType
         self.agentConfig = agentConfig  # type: AgentConfigType
@@ -299,6 +326,7 @@ def __init__(self, *args, **kwargs):
 
         self.__formatted_tags = None
         self.__logs_enabled = None
+        self.__security_config = None
         self.__persistent_cache_key_prefix: str = ""
 
         if os.environ.get("GOFIPS", "0") == "1":
@@ -401,6 +429,28 @@ def logs_enabled(self):
 
         return self.__logs_enabled
 
+    @property
+    def security_config(self) -> SecurityConfig:
+        """
+        Returns the integration security configuration, loaded once and cached.
+
+        The security config controls file path validation for untrusted providers.
+        """
+        if self.__security_config is None:
+            trusted_providers = datadog_agent.get_config('integration_trusted_providers')
+            self.__security_config = SecurityConfig(
+                check_name=self.name,
+                provider=self.provider,
+                ignore_untrusted_file_params=bool(datadog_agent.get_config('integration_ignore_untrusted_file_params')),
+                file_paths_allowlist=datadog_agent.get_config('integration_file_paths_allowlist') or [],
+                trusted_providers=trusted_providers
+                if trusted_providers is not None
+                else list(DEFAULT_TRUSTED_PROVIDERS),
+                excluded_checks=datadog_agent.get_config('integration_security_excluded_checks') or [],
+            )
+
+        return self.__security_config
+
     @property
     def formatted_tags(self):
         # type: () -> str
@@ -604,10 +654,27 @@ def load_configuration_model(import_path, model_name, config, context):
 
                 raise ConfigurationError('\n'.join(message_lines)) from None
             else:
+                # Fallback security validation for fields in GLOBAL_SECURE_FIELDS.
+                # This catches secure fields in integrations that haven't regenerated
+                # their models with require_trusted_provider in the spec.
+                try:
+                    security_config = context.get('security_config')
+                    configured_fields = context.get('configured_fields', frozenset())
+                    for field_name in GLOBAL_SECURE_FIELDS & configured_fields:
+                        value = config.get(field_name)
+                        if value is not None:
+                            check_field_trusted_provider(field_name, value, security_config)
+                except ValueError as e:
+                    raise ConfigurationError(str(e)) from None
                 return config_model
 
     def _get_config_model_context(self, config):
-        return {'logger': self.log, 'warning': self.warning, 'configured_fields': frozenset(config)}
+        return {
+            'logger': self.log,
+            'warning': self.warning,
+            'configured_fields': frozenset(config),
+            'security_config': self.security_config,
+        }
 
     def register_secret(self, secret: str) -> None:
         """

datadog_checks_base/datadog_checks/base/utils/models/validation/__init__.py

@@ -1,4 +1,4 @@
 # (C) Datadog, Inc. 2021-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
-from . import core, utils
+from . import core, security, utils

datadog_checks_base/datadog_checks/base/utils/models/validation/core.py

@@ -1,11 +1,16 @@
 # (C) Datadog, Inc. 2021-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
-def initialize_config(values, **kwargs):
+from __future__ import annotations
+
+from typing import Any
+
+
+def initialize_config(values: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
     # This is what is returned by the initial model validator of each config model.
     return values
 
 
-def check_model(model, **kwargs):
+def check_model(model: Any, **kwargs: Any) -> Any:
     # This is what is returned by the final model validator of each config model.
     return model

datadog_checks_base/datadog_checks/base/utils/models/validation/security.py

@@ -0,0 +1,92 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+
+DEFAULT_TRUSTED_PROVIDERS: tuple[str, ...] = ('file', 'remote-config')
+
+
+@dataclass
+class SecurityConfig:
+    """Security configuration for integration file path validation."""
+
+    check_name: str = ''
+    provider: str = ''
+    ignore_untrusted_file_params: bool = False
+    file_paths_allowlist: list[str] = field(default_factory=list)
+    trusted_providers: list[str] = field(default_factory=lambda: list(DEFAULT_TRUSTED_PROVIDERS))
+    excluded_checks: list[str] = field(default_factory=list)
+
+    def is_enabled(self) -> bool:
+        """Return whether file path security enforcement is enabled."""
+        return self.ignore_untrusted_file_params
+
+    def is_provider_trusted(self, provider: str) -> bool:
+        """Return whether the given provider is in the trusted providers list."""
+        return provider in self.trusted_providers
+
+    def is_file_path_allowed(self, path: str) -> bool:
+        """Return whether the resolved path falls under any allowed prefix directory."""
+        resolved = os.path.realpath(path)
+        return any(
+            resolved == os.path.realpath(allowed) or resolved.startswith(os.path.realpath(allowed) + os.sep)
+            for allowed in self.file_paths_allowlist
+        )
+
+    def is_check_excluded(self, check_name: str) -> bool:
+        """Return whether the given check is excluded from security restrictions."""
+        return check_name in self.excluded_checks
+
+
+def _get_auth_token_file_paths(value: dict) -> list[str]:
+    """Extract file paths from an auth_token object when reader.type is 'file'."""
+    reader = value.get('reader')
+    if not isinstance(reader, dict):
+        return []
+    if reader.get('type') != 'file':
+        return []
+    paths: list[str] = []
+    for key in ('path', 'private_key_path'):
+        path = reader.get(key)
+        if isinstance(path, str):
+            paths.append(path)
+    return paths
+
+
+def validate_require_trusted_provider(
+    field_name: str,
+    value: object,
+    security_config: SecurityConfig | None = None,
+) -> bool:
+    """Return True if the value is allowed based on security settings, False if it should be blocked."""
+    if security_config is None:
+        return True
+    if not security_config.is_enabled():
+        return True
+    if security_config.is_check_excluded(security_config.check_name):
+        return True
+    if security_config.is_provider_trusted(security_config.provider):
+        return True
+    if isinstance(value, str):
+        return security_config.is_file_path_allowed(value)
+    # auth_token is the only non-string field with require_trusted_provider;
+    # validate its reader.path when reader.type is 'file'
+    if field_name == 'auth_token' and isinstance(value, dict):
+        file_paths = _get_auth_token_file_paths(value)
+        if file_paths:
+            return all(security_config.is_file_path_allowed(p) for p in file_paths)
+        return True
+    return False
+
+
+def check_field_trusted_provider(
+    field_name: str,
+    value: object,
+    security_config: SecurityConfig | None,
+) -> None:
+    """Raise ValueError if the field value is not allowed from an untrusted provider."""
+    if not validate_require_trusted_provider(field_name, value, security_config):
+        raise ValueError(f"Field '{field_name}' is not allowed from untrusted provider '{security_config.provider}'")

datadog_checks_base/datadog_checks/base/utils/models/validation/utils.py

@@ -1,10 +1,13 @@
 # (C) Datadog, Inc. 2021-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
 from types import MappingProxyType
+from typing import Any, Callable
 
 
-def make_immutable(obj):
+def make_immutable(obj: Any) -> Any:
     if isinstance(obj, list):
         return tuple(make_immutable(item) for item in obj)
     elif isinstance(obj, dict):
@@ -13,8 +16,13 @@ def make_immutable(obj):
     return obj
 
 
-def handle_deprecations(config_section, deprecations, fields, context):
-    warning_method = context['warning']
+def handle_deprecations(
+    config_section: str,
+    deprecations: dict[str, dict[str, str]],
+    fields: set[str],
+    context: dict[str, Any],
+) -> None:
+    warning_method: Callable[..., Any] = context['warning']
 
     for option, data in deprecations.items():
         if option not in fields:

datadog_checks_base/tests/models/config_models/instance.py

@@ -20,6 +20,15 @@
 from . import defaults, deprecations, validators
 
 
+class AuthToken(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    reader: Optional[MappingProxyType[str, Any]] = None
+    writer: Optional[MappingProxyType[str, Any]] = None
+
+
 class Obj(BaseModel):
     model_config = ConfigDict(
         arbitrary_types_allowed=True,
@@ -36,6 +45,7 @@ class InstanceConfig(BaseModel):
         frozen=True,
     )
     array: Optional[tuple[str, ...]] = None
+    auth_token: Optional[AuthToken] = None
     deprecated: Optional[str] = None
     flag: Optional[bool] = None
     hyphenated_name: Optional[str] = Field(None, alias='hyphenated-name')
@@ -45,6 +55,7 @@ class InstanceConfig(BaseModel):
     pid: Optional[int] = None
     text: Optional[str] = None
     timeout: Optional[float] = None
+    tls_cert: Optional[str] = None
 
     @model_validator(mode='before')
     def _handle_deprecations(cls, values, info):

datadog_checks_base/tests/models/data/spec.yaml

@@ -76,3 +76,20 @@ files:
       description: words
       value:
         type: string
+    - name: tls_cert
+      description: Path to TLS certificate
+      value:
+        type: string
+        require_trusted_provider: true
+    - name: auth_token
+      description: Authentication token configuration
+      value:
+        type: object
+        require_trusted_provider: true
+        properties:
+          - name: reader
+            type: object
+            properties: []
+          - name: writer
+            type: object
+            properties: []