ddev: add trusted _fetch_command secret resolution to config fields

ddev/changelog.d/23050.added

@@ -0,0 +1 @@
+Add `*_fetch_command` secret resolution to ddev config fields, with trust-gating for `.ddev.toml` overrides and secret scrubbing in `config show`.

ddev/src/ddev/cli/__init__.py

@@ -130,6 +130,10 @@ def ddev(
         app.config_file.load()
     except OSError as e:  # no cov
         app.abort(f'Error loading configuration: {e}')
+
+    for warning in app.config_file.load_warnings:
+        app.display_warning(warning)
+
     if app.config.upgrade_check:
         upgrade_check.upgrade_check(app, __version__)
 

ddev/src/ddev/cli/config/__init__.py

@@ -4,6 +4,8 @@
 
 import click
 
+from ddev.cli.config.allow import allow
+from ddev.cli.config.deny import deny
 from ddev.cli.config.edit import edit
 from ddev.cli.config.explore import explore
 from ddev.cli.config.find import find
@@ -18,6 +20,8 @@ def config():
     pass
 
 
+config.add_command(allow)
+config.add_command(deny)
 config.add_command(edit)
 config.add_command(explore)
 config.add_command(find)

ddev/src/ddev/cli/config/allow.py

@@ -0,0 +1,33 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command(short_help='Trust the local .ddev.toml so its _fetch_command fields are executed')
+@click.pass_obj
+def allow(app: Application):
+    """Mark the local ``.ddev.toml`` as trusted.
+
+    When trusted, ``*_fetch_command`` fields in the override file are executed to
+    resolve secret values.  The current file hash is stored; if the file
+    changes the trust is automatically revoked and a warning is shown.
+    """
+    from ddev.config.override_trust import upsert_trust_entry
+
+    if not app.config_file.overrides_available():
+        app.abort(f'No {".ddev.toml"} file found in the current directory or any parent directory.')
+
+    upsert_trust_entry(
+        overrides_path=app.config_file.overrides_path,
+        global_config_dir=app.config_file.global_path.parent,
+        state='allowed',
+    )
+    app.display_success(f'Trusted: {app.config_file.pretty_overrides_path}')

ddev/src/ddev/cli/config/deny.py

@@ -0,0 +1,32 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command(short_help='Silence warnings about untrusted _fetch_command fields in .ddev.toml')
+@click.pass_obj
+def deny(app: Application):
+    """Mark the local ``.ddev.toml`` as explicitly untrusted.
+
+    ``*_fetch_command`` fields in the override file will be stripped silently
+    (no warning shown).  Use ``ddev config allow`` to re-enable execution.
+    """
+    from ddev.config.override_trust import upsert_trust_entry
+
+    if not app.config_file.overrides_available():
+        app.abort(f'No {".ddev.toml"} file found in the current directory or any parent directory.')
+
+    upsert_trust_entry(
+        overrides_path=app.config_file.overrides_path,
+        global_config_dir=app.config_file.global_path.parent,
+        state='denied',
+    )
+    app.display_success(f'Silenced: {app.config_file.pretty_overrides_path}')

ddev/src/ddev/cli/config/override.py

@@ -65,7 +65,7 @@ def override(app: Application):
     from rich.syntax import Syntax
 
     from ddev.config.file import DDEV_TOML, RootConfig, deep_merge_with_list_handling
-    from ddev.config.utils import scrub_config
+    from ddev.config.scrubber import scrub_config
     from ddev.utils.fs import Path
     from ddev.utils.toml import dumps_toml_data
 

ddev/src/ddev/cli/config/set.py

@@ -66,7 +66,8 @@ def set_value(app: Application, key: str, value: str | None, overrides: bool):
 
     import tomlkit
 
-    from ddev.config.utils import SCRUBBED_GLOBS, create_toml_document, save_toml_document, scrub_config
+    from ddev.config.scrubber import SCRUBBED_GLOBS, scrub_config
+    from ddev.config.utils import create_toml_document, save_toml_document
 
     scrubbing = any(fnmatch(key, glob) for glob in SCRUBBED_GLOBS)
     if value is None:

ddev/src/ddev/cli/meta/scripts/_dynamicd/cli.py

@@ -5,12 +5,12 @@
 
 from __future__ import annotations
 
-import os
 from typing import TYPE_CHECKING
 
 import click
 
 from ddev.cli.meta.scripts._dynamicd.constants import SCENARIOS
+from ddev.config.model import ConfigurationError
 
 if TYPE_CHECKING:
     from ddev.cli.application import Application
@@ -78,15 +78,19 @@ def _get_api_keys(app: Application) -> tuple[str, str]:
 
     Returns (llm_api_key, dd_api_key) or aborts if not configured.
     """
-    # Get LLM API key from config or environment variable
-    llm_api_key = app.config.raw_data.get("dynamicd", {}).get("llm_api_key")
-    if not llm_api_key:
-        llm_api_key = os.environ.get("ANTHROPIC_API_KEY")
+    # Resolve LLM API key lazily via the shared config model.
+    try:
+        llm_api_key = app.config.dynamicd.resolve_llm_api_key()
+    except ConfigurationError as e:
+        app.display_error(str(e))
+        app.abort()
+
     if not llm_api_key:
         app.display_error(
             "LLM API key not configured. Either:\n"
             "  1. Set env var: export ANTHROPIC_API_KEY=<your-key>\n"
-            "  2. Or run: ddev config set dynamicd.llm_api_key <your-key>"
+            "  2. Or run: ddev config set dynamicd.llm_api_key <your-key>\n"
+            "  3. Or run: ddev config set dynamicd.llm_api_key_fetch_command '<command>'"
         )
         app.abort()
 

ddev/src/ddev/config/command_resolver.py

@@ -0,0 +1,88 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+"""Per-process command resolver with deterministic caching for `*_fetch_command` secret fields."""
+
+from __future__ import annotations
+
+import subprocess
+
+# Failure reasons surfaced by CommandExecutionError.
+NON_ZERO_EXIT = 'non_zero_exit'
+EMPTY_OUTPUT = 'empty_output'
+
+# Per-process cache keyed by exact command string.
+_COMMAND_CACHE: dict[str, str] = {}
+
+
+class CommandExecutionError(Exception):
+    """Raised when a secret-resolution command exits non-zero or produces no output."""
+
+    _MAX_STDERR_CHARS = 200
+
+    def __init__(self, command: str, returncode: int, stderr: str, reason: str):
+        self.command = command
+        self.returncode = returncode
+        self.reason = reason
+        self.stderr = stderr.strip()
+        super().__init__(self._reason_message())
+
+    def _stderr_excerpt(self) -> str:
+        if not self.stderr:
+            return ''
+
+        # Keep user-facing output compact and single-line.
+        cleaned = ' '.join(self.stderr.split())
+        if len(cleaned) > self._MAX_STDERR_CHARS:
+            return f'{cleaned[: self._MAX_STDERR_CHARS - 3]}...'
+        return cleaned
+
+    def _reason_message(self) -> str:
+        if self.reason == EMPTY_OUTPUT:
+            return 'command returned empty output'
+
+        stderr_excerpt = self._stderr_excerpt()
+        if stderr_excerpt:
+            return f'command failed with exit code {self.returncode}: {stderr_excerpt}'
+        return f'command failed with exit code {self.returncode}'
+
+    def to_user_message(self, field_path: str) -> str:
+        return (
+            f"Failed to resolve `{field_path}`: {self._reason_message()}. "
+            "Check that the configured *_fetch_command exists, is executable, writes the secret to stdout, "
+            "and returns a non-empty value."
+        )
+
+
+def run_command(command: str) -> str:
+    """Execute *command* in a shell and return stdout stripped of surrounding whitespace.
+
+    Results are cached per-process so each distinct command string runs at most once.
+
+    Raises:
+        TypeError: if *command* is not a ``str``.
+        CommandExecutionError: if the command exits with a non-zero return code or
+            produces empty output after stripping.
+    """
+    if not isinstance(command, str):
+        raise TypeError(f'command must be a str, got {type(command).__name__!r}')
+
+    if command in _COMMAND_CACHE:
+        return _COMMAND_CACHE[command]
+
+    result = subprocess.run(command, shell=True, text=True, capture_output=True, check=False)
+
+    if result.returncode != 0:
+        raise CommandExecutionError(command, result.returncode, result.stderr, reason=NON_ZERO_EXIT)
+
+    value = result.stdout.strip()
+    if not value:
+        raise CommandExecutionError(command, result.returncode, result.stderr, reason=EMPTY_OUTPUT)
+
+    _COMMAND_CACHE[command] = value
+    return value
+
+
+def clear_cache() -> None:
+    """Clear the per-process command cache.  Intended for use in tests only."""
+    _COMMAND_CACHE.clear()

ddev/src/ddev/config/file.py

@@ -9,7 +9,9 @@
 from typing import cast
 
 from ddev.config.model import RootConfig
-from ddev.config.utils import scrub_config
+from ddev.config.override_trust import get_override_trust_state
+from ddev.config.override_trust import strip_fetch_command_fields as _strip_fetch_command_fields
+from ddev.config.scrubber import scrub_config
 from ddev.utils.fs import Path
 from ddev.utils.toml import dumps_toml_data, load_toml_data
 
@@ -94,6 +96,8 @@ def __init__(self, path: Path | None = None):
         self.combined_model: RootConfig = cast(RootConfig, UNINITIALIZED)
         self.combined_content: str = ""
         self._overrides_path: Path | None = None
+        # Warnings collected during load() for surfacing at CLI startup.
+        self.load_warnings: list[str] = []
 
     @property
     def overrides_path(self) -> Path:
@@ -146,6 +150,8 @@ def pretty_overrides_path(self) -> Path:
             return Path(relative_overrides_path)
 
     def load(self):
+        self.load_warnings = []
+
         self.global_content = self.global_path.read_text()
         self.global_model = RootConfig(load_toml_data(self.global_content))
 
@@ -157,7 +163,29 @@ def load(self):
             return
 
         self.overrides_content = overrides_content
-        self.overrides_model = RootConfig(load_toml_data(self.overrides_content))
+        overrides_raw = load_toml_data(self.overrides_content)
+
+        # Determine trust state for the local override file.
+        global_config_dir = self.global_path.parent
+        trust_state, _ = get_override_trust_state(self.overrides_path, global_config_dir)
+
+        if trust_state == 'allowed':
+            # Commands are trusted – leave overrides_raw intact.
+            pass
+        elif trust_state == 'denied':
+            # User explicitly silenced warnings; strip quietly.
+            _strip_fetch_command_fields(overrides_raw)
+        else:
+            # Unknown / hash mismatch – strip and warn.
+            stripped = _strip_fetch_command_fields(overrides_raw)
+            if stripped:
+                self.load_warnings.append(
+                    f'Ignored untrusted `_fetch_command` field(s) from {self.pretty_overrides_path}: '
+                    f'{", ".join(stripped)}. '
+                    f'Run `ddev config allow` to trust this file.'
+                )
+
+        self.overrides_model = RootConfig(overrides_raw)
 
         self.combined_model = RootConfig(
             deep_merge_with_list_handling(