feat(config): Add support for v1.1 protocol of secrets exec backend (#23655)

* feat(config): Add support for v1.1 of datadog secrets manager

* Update docs generator to generate unconstrained types

- This is necessary as the secrets exec config now contains a member of
type `Value`

* Introduce new unit tests for the exec backend

* Add changelog file

* Stray line to format

* Rename mock exec script and make it executable

- That way its implementation can change without modifying the
  executables call site in the code.

* Update src/secrets/exec.rs

* Install python 3.10 on windows CI runs

---------

Co-authored-by: Pavlos Rontidis <[email protected]>
Co-authored-by: Thomas <[email protected]>

Author: 3 people

.github/workflows/unit_windows.yml

@@ -7,7 +7,6 @@ permissions:
   statuses: write
 
 jobs:
-
   test-windows:
     runs-on: windows-2025-8core
     timeout-minutes: 60
@@ -31,6 +30,9 @@ jobs:
         if: ${{ github.event_name != 'pull_request_review' }}
         uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
 
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
       - run: .\scripts\environment\bootstrap-windows-2025.ps1
       - run: make test
 

changelog.d/add_support_v1_1_secrets.feature.md

@@ -0,0 +1,3 @@
+Add support for v1.1 of the secrets manager protocol
+
+authors: graphcareful

scripts/generate-component-docs.rb

@@ -928,6 +928,13 @@ def resolve_bare_schema(root_schema, schema)
       fix_grouped_enums_if_numeric!(grouped)
       grouped.transform_values! { |values| { 'enum' => values } }
       grouped
+    when nil
+      # Unconstrained/empty schema (e.g., Value without constraints).
+      # Represent it as accepting any JSON type so downstream code can render it
+      # and attach defaults/examples based on actual values.
+      @logger.debug 'Resolving unconstrained schema (any type).'
+
+      { '*' => {} }
     else
       @logger.error "Failed to resolve the schema. Schema: #{schema}"
       exit 1

src/secrets/exec.rs

@@ -7,9 +7,59 @@ use serde::{Deserialize, Serialize};
 use tokio::{io::AsyncWriteExt, process::Command, time};
 use tokio_util::codec;
 use vector_lib::configurable::{component::GenerateConfig, configurable_component};
+use vrl::value::Value;
 
 use crate::{config::SecretBackend, signal};
 
+/// Configuration for the command that will be `exec`ed
+#[configurable_component(secrets("exec"))]
+#[configurable(metadata(docs::enum_tag_description = "The protocol version."))]
+#[derive(Clone, Debug)]
+#[serde(rename_all = "snake_case", tag = "version")]
+pub enum ExecVersion {
+    /// Expect the command to fetch the configuration options itself.
+    V1,
+
+    /// Configuration options to the command are to be curried upon each request.
+    V1_1 {
+        /// The name of the backend. This is `type` field in the backend request.
+        backend_type: String,
+        /// The configuration to pass to the secrets executable. This is the `config` field in the
+        /// backend request. Refer to the documentation of your `backend_type `to see which options
+        /// are required to be set.
+        backend_config: Value,
+    },
+}
+
+impl ExecVersion {
+    fn new_query(&self, secrets: HashSet<String>) -> ExecQuery {
+        match &self {
+            ExecVersion::V1 => ExecQuery {
+                version: "1.0".to_string(),
+                secrets,
+                r#type: None,
+                config: None,
+            },
+            ExecVersion::V1_1 {
+                backend_type,
+                backend_config,
+                ..
+            } => ExecQuery {
+                version: "1.1".to_string(),
+                secrets,
+                r#type: Some(backend_type.clone()),
+                config: Some(backend_config.clone()),
+            },
+        }
+    }
+}
+
+impl GenerateConfig for ExecVersion {
+    fn generate_config() -> toml::Value {
+        toml::Value::try_from(ExecVersion::V1).unwrap()
+    }
+}
+
 /// Configuration for the `exec` secrets backend.
 #[configurable_component(secrets("exec"))]
 #[derive(Clone, Debug)]
@@ -22,13 +72,18 @@ pub struct ExecBackend {
     /// The timeout, in seconds, to wait for the command to complete.
     #[serde(default = "default_timeout_secs")]
     pub timeout: u64,
+
+    /// Settings for the protocol between Vector and the secrets executable.
+    #[serde(default = "default_protocol_version")]
+    pub protocol: ExecVersion,
 }
 
 impl GenerateConfig for ExecBackend {
     fn generate_config() -> toml::Value {
         toml::Value::try_from(ExecBackend {
             command: vec![String::from("/path/to/script")],
             timeout: 5,
+            protocol: ExecVersion::V1,
         })
         .unwrap()
     }
@@ -38,17 +93,20 @@ const fn default_timeout_secs() -> u64 {
     5
 }
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+const fn default_protocol_version() -> ExecVersion {
+    ExecVersion::V1
+}
+
+#[derive(Clone, Debug, Serialize)]
 struct ExecQuery {
+    // Fields in all versions starting from v1
     version: String,
     secrets: HashSet<String>,
-}
-
-fn new_query(secrets: HashSet<String>) -> ExecQuery {
-    ExecQuery {
-        version: "1.0".to_string(),
-        secrets,
-    }
+    // Fields added in v1.1
+    #[serde(skip_serializing_if = "Option::is_none")]
+    r#type: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    config: Option<Value>,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -66,7 +124,7 @@ impl SecretBackend for ExecBackend {
         let mut output = executor::block_on(async {
             query_backend(
                 &self.command,
-                new_query(secret_keys.clone()),
+                self.protocol.new_query(secret_keys.clone()),
                 self.timeout,
                 signal_rx,
             )
@@ -159,3 +217,77 @@ async fn query_backend(
     let response = serde_json::from_slice::<HashMap<String, ExecResponse>>(&output)?;
     Ok(response)
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        config::SecretBackend,
+        secrets::exec::{ExecBackend, ExecVersion},
+    };
+    use rstest::rstest;
+    use std::{
+        collections::{HashMap, HashSet},
+        path::PathBuf,
+    };
+    use tokio::sync::broadcast;
+    use vrl::value;
+
+    fn make_test_backend(protocol: ExecVersion) -> ExecBackend {
+        let command_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .join("tests/behavior/secrets/mock_secrets_exec");
+        ExecBackend {
+            command: [command_path.to_str().unwrap()].map(String::from).to_vec(),
+            timeout: 5,
+            protocol,
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    #[rstest(
+        protocol,
+        case(ExecVersion::V1),
+        case(ExecVersion::V1_1 {
+            backend_type: "file.json".to_string(),
+            backend_config: value!({"file_path": "/abc.json"}),
+        })
+    )]
+    async fn test_exec_backend(protocol: ExecVersion) {
+        let mut backend = make_test_backend(protocol);
+        let (_tx, mut rx) = broadcast::channel(1);
+        // These fake secrets are statically contained in mock_secrets_exec.py
+        let fake_secret_values: HashMap<String, String> = [
+            ("fake_secret_1", "123456"),
+            ("fake_secret_2", "123457"),
+            ("fake_secret_3", "123458"),
+            ("fake_secret_4", "123459"),
+            ("fake_secret_5", "123460"),
+        ]
+        .into_iter()
+        .map(|(k, v)| (k.to_string(), v.to_string()))
+        .collect();
+        // Calling the mock_secrets_exec.py program with the expected secret keys should provide
+        // the values expected above in `fake_secret_values`
+        let fetched_keys = backend
+            .retrieve(fake_secret_values.keys().cloned().collect(), &mut rx)
+            .await
+            .unwrap();
+        // Assert response is as expected
+        assert_eq!(fetched_keys.len(), 5);
+        for (fake_secret_key, fake_secret_value) in fake_secret_values {
+            assert_eq!(fetched_keys.get(&fake_secret_key), Some(&fake_secret_value));
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_exec_backend_missing_secrets() {
+        let mut backend = make_test_backend(ExecVersion::V1);
+        let (_tx, mut rx) = broadcast::channel(1);
+        let query_secrets: HashSet<String> =
+            ["fake_secret_900"].into_iter().map(String::from).collect();
+        let fetched_keys = backend.retrieve(query_secrets.clone(), &mut rx).await;
+        assert_eq!(
+            format!("{}", fetched_keys.unwrap_err()),
+            "secret for key 'fake_secret_900' was not retrieved: backend does not provide secret key"
+        );
+    }
+}

tests/behavior/secrets/mock_secrets_exec

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+
+"""
+Mock secret manager implementation used for testing the secrets.exec backend type.
+This program is meant to be exec'ed by a unit test so that the implementation can be tested.
+"""
+
+import sys
+import json
+from typing import Any
+
+
+class Request:
+    def __init__(
+        self,
+        version: str,
+        secrets: list[str],
+        type: str | None = None,
+        config: dict[str, Any] | None = None,
+    ):
+        self.version = version
+        self.secrets = secrets
+        self.type = type
+        self.config = config
+
+
+class Response:
+    def __init__(self, contents: dict[str, dict[str, str | None]]):
+        self.contents = contents
+
+
+def parse_request(req: dict[str, Any]) -> Request:
+    """
+    Validate the request by ensuring the correct keys exist per version, and that the types of the
+    respective keys are as expected as well
+    """
+    v1_args = set(["version", "secrets"])
+    v1_1_args = set(["version", "secrets", "type", "config"])
+    if "version" not in req.keys():
+        raise RuntimeError("version key missing from request")
+    version = req["version"]
+    if version == "1.0":
+        if v1_args != set(req.keys()):
+            raise RuntimeError(f"Invalid required keys in 1.0 request: {req.keys()}")
+        if not isinstance(req["secrets"], list):
+            raise RuntimeError("key 'secrets' should be a list")
+        return Request(version, req["secrets"])
+    elif version == "1.1":
+        if v1_1_args != set(req.keys()):
+            raise RuntimeError(f"Invalid required keys in 1.1 request: {req.keys()}")
+        if not isinstance(req["secrets"], list):
+            raise RuntimeError("key 'secrets' should be a list")
+        if not isinstance(req["type"], str):
+            raise RuntimeError("key 'type' should be a str")
+        if not isinstance(req["config"], dict):
+            raise RuntimeError("key 'config' should be a dict")
+        return Request(version, req["secrets"], req["type"], req["config"])
+    else:
+        raise RuntimeError(f"Invalid version detected: {version}")
+
+
+def handle_request(req: Request) -> Response:
+    """
+    Handle the request by looking up the requested secret with a value that is in a static fake
+    secret cache below. Any values not contained will return the appropriate error message.
+    """
+    static_fake_secrets_cache = {
+        "fake_secret_1": "123456",
+        "fake_secret_2": "123457",
+        "fake_secret_3": "123458",
+        "fake_secret_4": "123459",
+        "fake_secret_5": "123460",
+    }
+    supported_fake_backends = ["file.json"]
+    if req.version == "1.1":
+        if req.type not in supported_fake_backends:
+            raise RuntimeError(f"Requested backend: {req.type} not supported")
+        if req.config is not None and "file_path" not in req.config:
+            raise RuntimeError("File backend option file_path must be supplied")
+
+    def get_secret(fake_secret_name: str) -> dict[str, str | None]:
+        if fake_secret_name in static_fake_secrets_cache:
+            return {"value": static_fake_secrets_cache[fake_secret_name], "error": None}
+        else:
+            return {"value": None, "error": "backend does not provide secret key"}
+
+    return Response(dict([(s, get_secret(s)) for s in req.secrets]))
+
+
+def main():
+    data = sys.stdin.buffer.read()
+    req = json.loads(data)
+    req = parse_request(req)
+    resp = handle_request(req)
+    sys.stdout.write(json.dumps(resp.contents))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        sys.stderr.write(str(e))
+        sys.exit(1)